refactoring

extract shared logic to abstract class rewrite python transformer

Author: seko0716

python-transformer/src/main/java/seko/kafka/connect/transformer/python/PythonTransformer.java

@@ -1,56 +1,23 @@
 package seko.kafka.connect.transformer.python;
 
-import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.connect.connector.ConnectRecord;
-import org.apache.kafka.connect.transforms.Transformation;
-import org.apache.kafka.connect.transforms.util.Requirements;
-import org.apache.kafka.connect.transforms.util.SimpleConfig;
 import org.python.core.PyCode;
 import org.python.core.PyObject;
 import org.python.util.PythonInterpreter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import seko.kafka.connect.transformer.script.AbstractScriptTransformer;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 
-import static org.apache.kafka.common.config.ConfigDef.Importance.MEDIUM;
-import static org.apache.kafka.common.config.ConfigDef.NO_DEFAULT_VALUE;
-import static org.apache.kafka.common.config.ConfigDef.Type.STRING;
-import static seko.kafka.connect.transformer.script.configs.Configuration.KEY_SCRIPT_CONFIG;
-import static seko.kafka.connect.transformer.script.configs.Configuration.VALUE_SCRIPT_CONFIG;
-
-public class PythonTransformer<R extends ConnectRecord<R>> implements Transformation<R> {
- private static final String PURPOSE = "field extraction";
- private static final ConfigDef CONFIG_DEF = new ConfigDef()
- .define(KEY_SCRIPT_CONFIG, STRING, NO_DEFAULT_VALUE, MEDIUM, "Field name to extract.")
- .define(VALUE_SCRIPT_CONFIG, STRING, NO_DEFAULT_VALUE, MEDIUM, "Format extracted field.");
-
-
+public class PythonTransformer<R extends ConnectRecord<R>> extends AbstractScriptTransformer {
  private static final Logger log = LoggerFactory.getLogger(PythonTransformer.class);
 
- private String valueScript;
- private String keyScript;
-
- @Override
- public R apply(R record) {
- Map<String, Object> key = Requirements.requireMapOrNull(record.key(), PURPOSE);
-
- if (keyScript != null && key != null) {
- key = transform(key, keyScript);
- }
 
- Map<String, Object> value = Requirements.requireMapOrNull(record.value(), PURPOSE);
- if (valueScript != null && value != null) {
- value = transform(value, valueScript);
- }
-
- return newRecord(record, key, value);
- }
-
- private Map<String, Object> transform(Map<String, Object> source, String script) {
+ public Map<String, Object> transform(Map<String, Object> source, String script) {
  PythonInterpreter interpreter = new PythonInterpreter();
  interpreter.set("source", source);
 
@@ -69,35 +36,4 @@ private Map<String, Object> transform(Map<String, Object> source, String script)
  }
  }
 
- @Override
- public ConfigDef config() {
- return CONFIG_DEF;
- }
-
-
- private R newRecord(R record, Map<String, Object> newKey, Map<String, Object> newValue) {
- Object key = newKey == null ? record.key() : newKey;
- Object value = newValue == null ? record.value() : newValue;
-
- return record.newRecord(record.topic(), record.kafkaPartition(), record.keySchema(),
- key, record.valueSchema(), value, record.timestamp());
- }
-
- @Override
- public void close() {
-
- }
-
- @Override
- public void configure(Map<String, ?> configs) {
- SimpleConfig config = new SimpleConfig(CONFIG_DEF, configs);
- String keyScript = config.getString(KEY_SCRIPT_CONFIG);
- if (keyScript != null && !keyScript.trim().isEmpty()) {
- this.keyScript = keyScript;
- }
- String valueScript = config.getString(VALUE_SCRIPT_CONFIG);
- if (valueScript != null && !valueScript.trim().isEmpty()) {
- this.valueScript = valueScript;
- }
- }
 }

python-transformer/src/test/java/seko/kafka/connect/transformer/python/PythonTransformerTest.java

@@ -1,5 +1,6 @@
 package seko.kafka.connect.transformer.python;
 
+import org.apache.kafka.connect.connector.ConnectRecord;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.apache.kafka.connect.transforms.util.Requirements;
 import org.junit.Assert;
@@ -27,7 +28,7 @@ public void applyWithoutSchemaString() {
  Map<String, Object> event = new HashMap<>();
  event.put("created_when", "2019-05-31T00:17:00.188Z");
 
- SourceRecord transformed = dateRouter.apply(new SourceRecord(null, null, "topic", 0, null, event));
+ ConnectRecord transformed = dateRouter.apply(new SourceRecord(null, null, "topic", 0, null, event));
  Map<String, Object> stringObjectMap = Requirements.requireMapOrNull(transformed.value(), "");
  Assert.assertEquals(12312312, stringObjectMap.get("qweqweq"));
  }