Allow providing custom Kafka properties

RuckusWirelessIL · Jan 5, 2017 · 10fe128 · 10fe128
1 parent c2709cf
commit 10fe128
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 14 deletions.
diff --git a/src/main/java/com/ruckuswireless/pentaho/kafka/producer/KafkaProducerDialog.java b/src/main/java/com/ruckuswireless/pentaho/kafka/producer/KafkaProducerDialog.java
@@ -1,6 +1,8 @@
 package com.ruckuswireless.pentaho.kafka.producer;
 
+import java.util.Arrays;
 import java.util.Properties;
+import java.util.TreeSet;
 
 import org.eclipse.swt.SWT;
 import org.eclipse.swt.custom.CCombo;
@@ -256,11 +258,15 @@ private void getData(KafkaProducerMeta producerMeta, boolean copyStepname) {
 		wMessageField.setText(Const.NVL(producerMeta.getMessageField(), ""));
 		wKeyField.setText(Const.NVL(producerMeta.getKeyField(), ""));
 
+		TreeSet<String> propNames = new TreeSet<String>();
+		propNames.addAll(Arrays.asList(KafkaProducerMeta.KAFKA_PROPERTIES_NAMES));
+		propNames.addAll(producerMeta.getKafkaProperties().stringPropertyNames());
+
 		Properties kafkaProperties = producerMeta.getKafkaProperties();
-		for (int i = 0; i < KafkaProducerMeta.KAFKA_PROPERTIES_NAMES.length; ++i) {
-			String propName = KafkaProducerMeta.KAFKA_PROPERTIES_NAMES[i];
+		int i = 0;
+		for (String propName : propNames) {
 			String value = kafkaProperties.getProperty(propName);
-			TableItem item = new TableItem(wProps.table, i > 1 ? SWT.BOLD : SWT.NONE);
+			TableItem item = new TableItem(wProps.table, i++ > 1 ? SWT.BOLD : SWT.NONE);
 			int colnr = 1;
 			item.setText(colnr++, Const.NVL(propName, ""));
 			String defaultValue = KafkaProducerMeta.KAFKA_PROPERTIES_DEFAULTS.get(propName);
@@ -269,6 +275,7 @@ private void getData(KafkaProducerMeta producerMeta, boolean copyStepname) {
 			}
 			item.setText(colnr++, Const.NVL(value, defaultValue));
 		}
+
 		wProps.removeEmptyRows();
 		wProps.setRowNums();
 		wProps.optWidth(true);

diff --git a/src/main/java/com/ruckuswireless/pentaho/kafka/producer/KafkaProducerMeta.java b/src/main/java/com/ruckuswireless/pentaho/kafka/producer/KafkaProducerMeta.java
@@ -1,5 +1,7 @@
 package com.ruckuswireless.pentaho.kafka.producer;
 
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -138,10 +140,13 @@ public void loadXML(Node stepnode, List<DatabaseMeta> databases, Map<String, Cou
 			messageField = XMLHandler.getTagValue(stepnode, "FIELD");
 			keyField = XMLHandler.getTagValue(stepnode, "KEYFIELD");
 			Node kafkaNode = XMLHandler.getSubNode(stepnode, "KAFKA");
-			for (String name : KAFKA_PROPERTIES_NAMES) {
-				String value = XMLHandler.getTagValue(kafkaNode, name);
-				if (value != null) {
-					kafkaProperties.put(name, value);
+			String[] kafkaElements = XMLHandler.getNodeElements(kafkaNode);
+			if (kafkaElements != null) {
+				for (String propName : kafkaElements) {
+					String value = XMLHandler.getTagValue(kafkaNode, propName);
+					if (value != null) {
+						kafkaProperties.put(propName, value);
+					}
 				}
 			}
 		} catch (Exception e) {
@@ -161,7 +166,7 @@ public String getXML() throws KettleException {
 			retval.append("    ").append(XMLHandler.addTagValue("KEYFIELD", keyField));
 		}
 		retval.append("    ").append(XMLHandler.openTag("KAFKA")).append(Const.CR);
-		for (String name : KAFKA_PROPERTIES_NAMES) {
+		for (String name : kafkaProperties.stringPropertyNames()) {
 			String value = kafkaProperties.getProperty(name);
 			if (value != null) {
 				retval.append("      " + XMLHandler.addTagValue(name, value));
@@ -177,6 +182,11 @@ public void readRep(Repository rep, ObjectId stepId, List<DatabaseMeta> database
 			topic = rep.getStepAttributeString(stepId, "TOPIC");
 			messageField = rep.getStepAttributeString(stepId, "FIELD");
 			keyField = rep.getStepAttributeString(stepId, "KEYFIELD");
+			String kafkaPropsXML = rep.getStepAttributeString(stepId, "KAFKA");
+			if (kafkaPropsXML != null) {
+				kafkaProperties.loadFromXML(new ByteArrayInputStream(kafkaPropsXML.getBytes()));
+			}
+			// Support old versions:
 			for (String name : KAFKA_PROPERTIES_NAMES) {
 				String value = rep.getStepAttributeString(stepId, name);
 				if (value != null) {
@@ -199,12 +209,9 @@ public void saveRep(Repository rep, ObjectId transformationId, ObjectId stepId)
 			if (keyField != null) {
 				rep.saveStepAttribute(transformationId, stepId, "KEYFIELD", keyField);
 			}
-			for (String name : KAFKA_PROPERTIES_NAMES) {
-				String value = kafkaProperties.getProperty(name);
-				if (value != null) {
-					rep.saveStepAttribute(transformationId, stepId, name, value);
-				}
-			}
+			ByteArrayOutputStream buf = new ByteArrayOutputStream();
+			kafkaProperties.storeToXML(buf, null);
+			rep.saveStepAttribute(transformationId, stepId, "KAFKA", buf.toString());
 		} catch (Exception e) {
 			throw new KettleException("KafkaProducerMeta.Exception.saveRep", e);
 		}