Seperated out Configuration into source and sink where appropriate
Added a few more tests
diff --git a/build.gradle b/build.gradle
index a131d22..114b3cd 100644
--- a/build.gradle
+++ b/build.gradle
@@ -19,7 +19,6 @@
compile group: 'org.apache.logging.log4j', name: 'log4j-api', version: '2.13.0'
compile group: 'org.apache.logging.log4j', name: 'log4j-core', version: '2.13.0'
-
testCompile(group: 'org.apache.kafka', name: 'kafka_2.12', version: '2.3.1')
testCompile(group: 'org.apache.kafka', name: 'kafka-streams-test-utils', version: '1.1.0')
testCompile(group: 'org.apache.curator', name: 'curator-framework', version: '4.2.0')
@@ -31,3 +30,4 @@
testImplementation 'org.awaitility:awaitility:4.0.2'
}
+
diff --git a/settings.gradle b/settings.gradle
index b5a2326..48eed3b 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -1,2 +1 @@
-rootProject.name = 'geode-kafka-connector'
-
+rootProject.name = 'geode-kafka-connector'
\ No newline at end of file
diff --git a/src/main/java/geode/kafka/GeodeConnectorConfig.java b/src/main/java/geode/kafka/GeodeConnectorConfig.java
index dcc479e..396d428 100644
--- a/src/main/java/geode/kafka/GeodeConnectorConfig.java
+++ b/src/main/java/geode/kafka/GeodeConnectorConfig.java
@@ -8,79 +8,26 @@
public class GeodeConnectorConfig {
- //Geode Configuration
- public static final String DURABLE_CLIENT_ID_PREFIX = "durableClientId";
- public static final String DEFAULT_DURABLE_CLIENT_ID = "";
- public static final String DURABLE_CLIENT_TIME_OUT = "durableClientTimeout";
- public static final String DEFAULT_DURABLE_CLIENT_TIMEOUT = "60000";
-
//GeodeKafka Specific Configuration
+ /**
+ * Identifier for each task
+ */
public static final String TASK_ID = "GEODE_TASK_ID"; //One config per task
-
- public static final String CQ_PREFIX = "cqPrefix";
- public static final String DEFAULT_CQ_PREFIX = "cqForGeodeKafka";
/**
* Specifies which Locators to connect to Apache Geode
*/
public static final String LOCATORS = "locators";
public static final String DEFAULT_LOCATOR = "localhost[10334]";
+ protected final int taskId;
+ protected List<LocatorHostPort> locatorHostPorts;
- /**
- * Specifies which Topics to connect in Kafka, uses the variable name with Kafka Sink Configuration
- * Only used in sink configuration
- */
- public static final String TOPICS = "topics";
-
- //Used by sink
- public static final String TOPIC_TO_REGION_BINDINGS = "topicToRegion";
-
- //Used by source
- public static final String REGION_TO_TOPIC_BINDINGS = "regionToTopic";
-
- /**
- * Property to describe the Source Partition in a record
- */
- public static final String REGION_NAME = "regionName"; //used for Source Partition Events
-
- public static final String BATCH_SIZE = "geodeConnectorBatchSize";
- public static final String DEFAULT_BATCH_SIZE = "100";
-
- public static final String QUEUE_SIZE = "geodeConnectorQueueSize";
- public static final String DEFAULT_QUEUE_SIZE = "100000";
-
- public static final String LOAD_ENTIRE_REGION = "loadEntireRegion";
- public static final String DEFAULT_LOAD_ENTIRE_REGION = "false";
-
-
- private final int taskId;
- private final String durableClientId;
- private final String durableClientIdPrefix;
- private final String durableClientTimeout;
-
- private Map<String, List<String>> regionToTopics;
- private Map<String, List<String>> topicToRegions;
- private List<LocatorHostPort> locatorHostPorts;
-
- //just for tests
- GeodeConnectorConfig() {
+ protected GeodeConnectorConfig() {
taskId = 0;
- durableClientId = "";
- durableClientIdPrefix = "";
- durableClientTimeout = "0";
}
public GeodeConnectorConfig(Map<String, String> connectorProperties) {
taskId = Integer.parseInt(connectorProperties.get(TASK_ID));
- durableClientIdPrefix = connectorProperties.get(DURABLE_CLIENT_ID_PREFIX);
- if (isDurable(durableClientIdPrefix)) {
- durableClientId = durableClientIdPrefix + taskId;
- } else {
- durableClientId = "";
- }
- durableClientTimeout = connectorProperties.get(DURABLE_CLIENT_TIME_OUT);
- regionToTopics = parseRegionToTopics(connectorProperties.get(GeodeConnectorConfig.REGION_TO_TOPIC_BINDINGS));
- topicToRegions = parseTopicToRegions(connectorProperties.get(GeodeConnectorConfig.TOPIC_TO_REGION_BINDINGS));
locatorHostPorts = parseLocators(connectorProperties.get(GeodeConnectorConfig.LOCATORS));
}
@@ -105,7 +52,7 @@
return bindings.stream().map(binding -> {
String[] regionToTopicsArray = parseBinding(binding);
return regionToTopicsArray;
- }).collect(Collectors.toMap(regionToTopicsArray -> regionToTopicsArray[0], regionToTopicsArray -> parseNames(regionToTopicsArray[1])));
+ }).collect(Collectors.toMap(regionToTopicsArray -> regionToTopicsArray[0], regionToTopicsArray -> parseStringByComma(regionToTopicsArray[1])));
}
public static List<String> parseBindings(String bindings) {
@@ -122,8 +69,12 @@
}
//Used to parse a string of topics or regions
- public static List<String> parseNames(String names) {
- return Arrays.stream(names.split(",")).map((s) -> s.trim()).collect(Collectors.toList());
+ public static List<String> parseStringByComma(String string) {
+ return parseStringBy(string, ",");
+ }
+
+ public static List<String> parseStringBy(String string, String regex) {
+ return Arrays.stream(string.split(regex)).map((s) -> s.trim()).collect(Collectors.toList());
}
public static String reconstructString(Collection<String> strings) {
@@ -144,40 +95,11 @@
return new LocatorHostPort(locator, port);
}
- public boolean isDurable() {
- return isDurable(durableClientId);
- }
-
- /**
- * @param durableClientId or prefix can be passed in. Either both will be "" or both will have a value
- * @return
- */
- boolean isDurable(String durableClientId) {
- return !durableClientId.equals("");
- }
-
public int getTaskId() {
return taskId;
}
- public String getDurableClientId() {
- return durableClientId;
- }
-
- public String getDurableClientTimeout() {
- return durableClientTimeout;
- }
-
public List<LocatorHostPort> getLocatorHostPorts() {
return locatorHostPorts;
}
-
- public Map<String, List<String>> getRegionToTopics() {
- return regionToTopics;
- }
-
- public Map<String, List<String>> getTopicToRegions() {
- return topicToRegions;
- }
-
}
diff --git a/src/main/java/geode/kafka/GeodeContext.java b/src/main/java/geode/kafka/GeodeContext.java
index d1fd3ae..b7c3d27 100644
--- a/src/main/java/geode/kafka/GeodeContext.java
+++ b/src/main/java/geode/kafka/GeodeContext.java
@@ -2,6 +2,8 @@
import org.apache.geode.cache.client.ClientCache;
import org.apache.geode.cache.client.ClientCacheFactory;
+import org.apache.geode.cache.client.PoolFactory;
+import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.query.CqAttributes;
import org.apache.geode.cache.query.CqException;
import org.apache.geode.cache.query.CqExistsException;
@@ -17,20 +19,42 @@
private ClientCache clientCache;
- public GeodeContext(GeodeConnectorConfig connectorConfig) {
- clientCache = createClientCache(connectorConfig.getLocatorHostPorts(), connectorConfig.getDurableClientId(), connectorConfig.getDurableClientTimeout());
+ public GeodeContext() {
+ }
+
+ public ClientCache connectClient(List<LocatorHostPort> locatorHostPortList, String durableClientId, String durableClientTimeout) {
+ clientCache = createClientCache(locatorHostPortList, durableClientId, durableClientTimeout);
+ return clientCache;
+ }
+
+ public ClientCache connectClient(List<LocatorHostPort> locatorHostPortList) {
+ clientCache = createClientCache(locatorHostPortList, "", "");
+ return clientCache;
}
public ClientCache getClientCache() {
return clientCache;
}
+ /**
+ *
+ * @param locators
+ * @param durableClientName
+ * @param durableClientTimeOut
+ * @return
+ */
public ClientCache createClientCache(List<LocatorHostPort> locators, String durableClientName, String durableClientTimeOut) {
- ClientCacheFactory ccf = new ClientCacheFactory().set("durable-client-id", durableClientName)
- .set("durable-client-timeout", durableClientTimeOut)
- .setPoolSubscriptionEnabled(true);
+ ClientCacheFactory ccf = new ClientCacheFactory();
+ if (!durableClientName.equals("")) {
+ ccf.set("durable-client-id", durableClientName)
+ .set("durable-client-timeout", durableClientTimeOut);
+ }
+ //currently we only allow using the default pool.
+ //If we ever want to allow adding multiple pools we'll have to configure pool factories
+ ccf.setPoolSubscriptionEnabled(true);
+
for (LocatorHostPort locator: locators) {
- ccf.addPoolLocator(locator.getHostName(), locator.getPort()).create();
+ ccf.addPoolLocator(locator.getHostName(), locator.getPort());
}
return ccf.create();
}
diff --git a/src/main/java/geode/kafka/sink/BatchRecords.java b/src/main/java/geode/kafka/sink/BatchRecords.java
index 742dcbc..282ba1c 100644
--- a/src/main/java/geode/kafka/sink/BatchRecords.java
+++ b/src/main/java/geode/kafka/sink/BatchRecords.java
@@ -4,6 +4,7 @@
import org.apache.kafka.connect.sink.SinkRecord;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -13,8 +14,18 @@
*/
public class BatchRecords {
- private Map updateMap = new HashMap();
- private List removeList = new ArrayList();
+ private Map updateMap;
+ private Collection removeList;
+
+ public BatchRecords() {
+ this(new HashMap(), new ArrayList());
+ }
+
+ /** Used for tests**/
+ public BatchRecords(Map updateMap, Collection removeList) {
+ this.updateMap = updateMap;
+ this.removeList = removeList;
+ }
public void addRemoveOperation(SinkRecord record) {
//if a previous operation added to the update map
diff --git a/src/main/java/geode/kafka/sink/GeodeKafkaSink.java b/src/main/java/geode/kafka/sink/GeodeKafkaSink.java
index 99f9b9d..43f8eab 100644
--- a/src/main/java/geode/kafka/sink/GeodeKafkaSink.java
+++ b/src/main/java/geode/kafka/sink/GeodeKafkaSink.java
@@ -11,19 +11,11 @@
import java.util.List;
import java.util.Map;
-import static geode.kafka.GeodeConnectorConfig.BATCH_SIZE;
-import static geode.kafka.GeodeConnectorConfig.CQ_PREFIX;
-import static geode.kafka.GeodeConnectorConfig.DEFAULT_BATCH_SIZE;
-import static geode.kafka.GeodeConnectorConfig.DEFAULT_CQ_PREFIX;
-import static geode.kafka.GeodeConnectorConfig.DEFAULT_DURABLE_CLIENT_ID;
-import static geode.kafka.GeodeConnectorConfig.DEFAULT_DURABLE_CLIENT_TIMEOUT;
import static geode.kafka.GeodeConnectorConfig.DEFAULT_LOCATOR;
-import static geode.kafka.GeodeConnectorConfig.DEFAULT_QUEUE_SIZE;
-import static geode.kafka.GeodeConnectorConfig.DURABLE_CLIENT_ID_PREFIX;
-import static geode.kafka.GeodeConnectorConfig.DURABLE_CLIENT_TIME_OUT;
import static geode.kafka.GeodeConnectorConfig.LOCATORS;
-import static geode.kafka.GeodeConnectorConfig.QUEUE_SIZE;
-import static geode.kafka.GeodeConnectorConfig.TOPIC_TO_REGION_BINDINGS;
+import static geode.kafka.GeodeSinkConnectorConfig.DEFAULT_NULL_VALUES_MEAN_REMOVE;
+import static geode.kafka.GeodeSinkConnectorConfig.NULL_VALUES_MEAN_REMOVE;
+import static geode.kafka.GeodeSinkConnectorConfig.TOPIC_TO_REGION_BINDINGS;
public class GeodeKafkaSink extends SinkConnector {
private static final ConfigDef CONFIG_DEF = new ConfigDef();
@@ -44,7 +36,7 @@
List<Map<String, String>> taskConfigs = new ArrayList<>();
Map<String, String> taskProps = new HashMap<>();
taskProps.putAll(sharedProps);
- List<String> bindings = GeodeConnectorConfig.parseNames(taskProps.get(TOPIC_TO_REGION_BINDINGS));
+ List<String> bindings = GeodeConnectorConfig.parseStringByComma(taskProps.get(TOPIC_TO_REGION_BINDINGS));
List<List<String>> bindingsPerTask = ConnectorUtils.groupPartitions(bindings, maxTasks);
for (int i = 0; i < maxTasks; i++) {
@@ -75,11 +67,7 @@
private Map<String, String> computeMissingConfigurations(Map<String, String> props) {
props.computeIfAbsent(LOCATORS, (key)-> DEFAULT_LOCATOR);
- props.computeIfAbsent(DURABLE_CLIENT_TIME_OUT, (key) -> DEFAULT_DURABLE_CLIENT_TIMEOUT);
- props.computeIfAbsent(DURABLE_CLIENT_ID_PREFIX, (key) -> DEFAULT_DURABLE_CLIENT_ID);
- props.computeIfAbsent(BATCH_SIZE, (key) -> DEFAULT_BATCH_SIZE);
- props.computeIfAbsent(QUEUE_SIZE, (key) -> DEFAULT_QUEUE_SIZE);
- props.computeIfAbsent(CQ_PREFIX, (key) -> DEFAULT_CQ_PREFIX);
+ props.computeIfAbsent(NULL_VALUES_MEAN_REMOVE, (key) -> DEFAULT_NULL_VALUES_MEAN_REMOVE);
return props;
}
}
diff --git a/src/main/java/geode/kafka/sink/GeodeKafkaSinkTask.java b/src/main/java/geode/kafka/sink/GeodeKafkaSinkTask.java
index 203192c..7f21134 100644
--- a/src/main/java/geode/kafka/sink/GeodeKafkaSinkTask.java
+++ b/src/main/java/geode/kafka/sink/GeodeKafkaSinkTask.java
@@ -1,7 +1,7 @@
package geode.kafka.sink;
-import geode.kafka.GeodeConnectorConfig;
import geode.kafka.GeodeContext;
+import geode.kafka.GeodeSinkConnectorConfig;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.client.ClientRegionShortcut;
import org.apache.kafka.connect.sink.SinkRecord;
@@ -9,13 +9,13 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
+
/**
* TODO javaDoc
* Currently force 1 region per task
@@ -24,10 +24,10 @@
private static final Logger logger = LoggerFactory.getLogger(GeodeKafkaSinkTask.class);
- GeodeContext geodeContext;
- Map<String, List<String>> topicToRegions;
- Map<String, Region> regionNameToRegion;
- boolean nullValuesMeansRemove = true;
+ private GeodeContext geodeContext;
+ private Map<String, List<String>> topicToRegions;
+ private Map<String, Region> regionNameToRegion;
+ private boolean nullValuesMeansRemove = true;
/**
* {@inheritDoc}
@@ -41,11 +41,13 @@
@Override
public void start(Map<String, String> props) {
try {
- GeodeConnectorConfig geodeConnectorConfig = new GeodeConnectorConfig(props);
+ GeodeSinkConnectorConfig geodeConnectorConfig = new GeodeSinkConnectorConfig(props);
logger.debug("GeodeKafkaSourceTask id:" + geodeConnectorConfig.getTaskId() + " starting");
- geodeContext = new GeodeContext(geodeConnectorConfig);
+ geodeContext = new GeodeContext();
+ geodeContext.connectClient(geodeConnectorConfig.getLocatorHostPorts());
topicToRegions = geodeConnectorConfig.getTopicToRegions();
regionNameToRegion = createProxyRegions(topicToRegions.values());
+ nullValuesMeansRemove = geodeConnectorConfig.getNullValuesMeanRemove();
} catch (Exception e) {
e.printStackTrace();
logger.error("Unable to start sink task", e);
diff --git a/src/main/java/geode/kafka/sink/GeodeSinkConnectorConfig.java b/src/main/java/geode/kafka/sink/GeodeSinkConnectorConfig.java
new file mode 100644
index 0000000..7c6aa3e
--- /dev/null
+++ b/src/main/java/geode/kafka/sink/GeodeSinkConnectorConfig.java
@@ -0,0 +1,35 @@
+package geode.kafka;
+
+import java.util.List;
+import java.util.Map;
+
+public class GeodeSinkConnectorConfig extends GeodeConnectorConfig {
+ //Used by sink
+ public static final String TOPIC_TO_REGION_BINDINGS = "topicToRegion";
+ public static final String NULL_VALUES_MEAN_REMOVE = "nullValuesMeanRemove";
+ public static final String DEFAULT_NULL_VALUES_MEAN_REMOVE = "true";
+
+ private Map<String, List<String>> topicToRegions;
+ private final boolean nullValuesMeanRemove;
+
+ //just for tests
+ GeodeSinkConnectorConfig() {
+ super();
+ nullValuesMeanRemove = Boolean.parseBoolean(DEFAULT_NULL_VALUES_MEAN_REMOVE);
+ }
+
+ public GeodeSinkConnectorConfig(Map<String, String> connectorProperties) {
+ super(connectorProperties);
+ topicToRegions = parseTopicToRegions(connectorProperties.get(TOPIC_TO_REGION_BINDINGS));
+ nullValuesMeanRemove = Boolean.parseBoolean(connectorProperties.get(NULL_VALUES_MEAN_REMOVE));
+ }
+
+ public Map<String, List<String>> getTopicToRegions() {
+ return topicToRegions;
+ }
+
+ public boolean getNullValuesMeanRemove() {
+ return nullValuesMeanRemove;
+ }
+
+}
diff --git a/src/main/java/geode/kafka/source/GeodeKafkaSource.java b/src/main/java/geode/kafka/source/GeodeKafkaSource.java
index 91e6203..686359e 100644
--- a/src/main/java/geode/kafka/source/GeodeKafkaSource.java
+++ b/src/main/java/geode/kafka/source/GeodeKafkaSource.java
@@ -12,21 +12,21 @@
import java.util.List;
import java.util.Map;
-import static geode.kafka.GeodeConnectorConfig.BATCH_SIZE;
-import static geode.kafka.GeodeConnectorConfig.CQ_PREFIX;
-import static geode.kafka.GeodeConnectorConfig.DEFAULT_BATCH_SIZE;
-import static geode.kafka.GeodeConnectorConfig.DEFAULT_CQ_PREFIX;
-import static geode.kafka.GeodeConnectorConfig.DEFAULT_DURABLE_CLIENT_ID;
-import static geode.kafka.GeodeConnectorConfig.DEFAULT_DURABLE_CLIENT_TIMEOUT;
-import static geode.kafka.GeodeConnectorConfig.DEFAULT_LOAD_ENTIRE_REGION;
import static geode.kafka.GeodeConnectorConfig.DEFAULT_LOCATOR;
-import static geode.kafka.GeodeConnectorConfig.DEFAULT_QUEUE_SIZE;
-import static geode.kafka.GeodeConnectorConfig.DURABLE_CLIENT_ID_PREFIX;
-import static geode.kafka.GeodeConnectorConfig.DURABLE_CLIENT_TIME_OUT;
-import static geode.kafka.GeodeConnectorConfig.LOAD_ENTIRE_REGION;
import static geode.kafka.GeodeConnectorConfig.LOCATORS;
-import static geode.kafka.GeodeConnectorConfig.QUEUE_SIZE;
-import static geode.kafka.GeodeConnectorConfig.REGION_TO_TOPIC_BINDINGS;
+import static geode.kafka.source.GeodeSourceConnectorConfig.BATCH_SIZE;
+import static geode.kafka.source.GeodeSourceConnectorConfig.CQ_PREFIX;
+import static geode.kafka.source.GeodeSourceConnectorConfig.DEFAULT_BATCH_SIZE;
+import static geode.kafka.source.GeodeSourceConnectorConfig.DEFAULT_CQ_PREFIX;
+import static geode.kafka.source.GeodeSourceConnectorConfig.DEFAULT_DURABLE_CLIENT_ID;
+import static geode.kafka.source.GeodeSourceConnectorConfig.DEFAULT_DURABLE_CLIENT_TIMEOUT;
+import static geode.kafka.source.GeodeSourceConnectorConfig.DEFAULT_LOAD_ENTIRE_REGION;
+import static geode.kafka.source.GeodeSourceConnectorConfig.DEFAULT_QUEUE_SIZE;
+import static geode.kafka.source.GeodeSourceConnectorConfig.DURABLE_CLIENT_ID_PREFIX;
+import static geode.kafka.source.GeodeSourceConnectorConfig.DURABLE_CLIENT_TIME_OUT;
+import static geode.kafka.source.GeodeSourceConnectorConfig.LOAD_ENTIRE_REGION;
+import static geode.kafka.source.GeodeSourceConnectorConfig.QUEUE_SIZE;
+import static geode.kafka.source.GeodeSourceConnectorConfig.REGION_TO_TOPIC_BINDINGS;
public class GeodeKafkaSource extends SourceConnector {
@@ -47,7 +47,7 @@
Map<String, String> taskProps = new HashMap<>();
taskProps.putAll(sharedProps);
- List<String> bindings = GeodeConnectorConfig.parseNames(taskProps.get(REGION_TO_TOPIC_BINDINGS));
+ List<String> bindings = GeodeConnectorConfig.parseStringByComma(taskProps.get(REGION_TO_TOPIC_BINDINGS));
List<List<String>> bindingsPerTask = ConnectorUtils.groupPartitions(bindings, maxTasks);
for (int i = 0; i < maxTasks; i++) {
diff --git a/src/main/java/geode/kafka/source/GeodeKafkaSourceTask.java b/src/main/java/geode/kafka/source/GeodeKafkaSourceTask.java
index dadc8ba..c983a51 100644
--- a/src/main/java/geode/kafka/source/GeodeKafkaSourceTask.java
+++ b/src/main/java/geode/kafka/source/GeodeKafkaSourceTask.java
@@ -19,12 +19,9 @@
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Collectors;
-import static geode.kafka.GeodeConnectorConfig.BATCH_SIZE;
-import static geode.kafka.GeodeConnectorConfig.CQ_PREFIX;
-import static geode.kafka.GeodeConnectorConfig.DEFAULT_BATCH_SIZE;
-import static geode.kafka.GeodeConnectorConfig.LOAD_ENTIRE_REGION;
-import static geode.kafka.GeodeConnectorConfig.QUEUE_SIZE;
-import static geode.kafka.GeodeConnectorConfig.REGION_NAME;
+import static geode.kafka.source.GeodeSourceConnectorConfig.BATCH_SIZE;
+import static geode.kafka.source.GeodeSourceConnectorConfig.QUEUE_SIZE;
+import static org.apache.geode.pdx.internal.PeerTypeRegistration.REGION_NAME;
public class GeodeKafkaSourceTask extends SourceTask {
@@ -37,7 +34,7 @@
private static final Map<String, Long> OFFSET_DEFAULT = createOffset();
private GeodeContext geodeContext;
- private GeodeConnectorConfig geodeConnectorConfig;
+ private GeodeSourceConnectorConfig geodeConnectorConfig;
private Map<String, List<String>> regionToTopics;
private Map<String, Map<String, String>> sourcePartitions;
private BlockingQueue<GeodeEvent> eventBuffer;
@@ -58,9 +55,11 @@
@Override
public void start(Map<String, String> props) {
try {
- geodeConnectorConfig = new GeodeConnectorConfig(props);
+ System.out.println("JASON start task");
+ geodeConnectorConfig = new GeodeSourceConnectorConfig(props);
logger.debug("GeodeKafkaSourceTask id:" + geodeConnectorConfig.getTaskId() + " starting");
- geodeContext = new GeodeContext(geodeConnectorConfig);
+ geodeContext = new GeodeContext();
+ geodeContext.connectClient(geodeConnectorConfig.getLocatorHostPorts(), geodeConnectorConfig.getDurableClientId(), geodeConnectorConfig.getDurableClientTimeout());
batchSize = Integer.parseInt(props.get(BATCH_SIZE));
int queueSize = Integer.parseInt(props.get(QUEUE_SIZE));
@@ -69,14 +68,18 @@
regionToTopics = geodeConnectorConfig.getRegionToTopics();
sourcePartitions = createSourcePartitionsMap(regionToTopics.keySet());
- String cqPrefix = props.get(CQ_PREFIX);
- boolean loadEntireRegion = Boolean.parseBoolean(props.get(LOAD_ENTIRE_REGION));
+ String cqPrefix = geodeConnectorConfig.getCqPrefix();
+ boolean loadEntireRegion = geodeConnectorConfig.getLoadEntireRegion();
installOnGeode(geodeConnectorConfig, geodeContext, eventBuffer, cqPrefix, loadEntireRegion);
} catch (Exception e) {
+ System.out.println("JASON start task failed" + e);
+ e.printStackTrace();
logger.error("Unable to start source task", e);
throw e;
}
+ System.out.println("JASON end task");
+
}
@Override
@@ -102,7 +105,7 @@
geodeContext.getClientCache().close(true);
}
- void installOnGeode(GeodeConnectorConfig geodeConnectorConfig, GeodeContext geodeContext, BlockingQueue eventBuffer, String cqPrefix, boolean loadEntireRegion) {
+ void installOnGeode(GeodeSourceConnectorConfig geodeConnectorConfig, GeodeContext geodeContext, BlockingQueue eventBuffer, String cqPrefix, boolean loadEntireRegion) {
boolean isDurable = geodeConnectorConfig.isDurable();
int taskId = geodeConnectorConfig.getTaskId();
for (String region : geodeConnectorConfig.getRegionToTopics().keySet()) {
diff --git a/src/main/java/geode/kafka/source/GeodeSourceConnectorConfig.java b/src/main/java/geode/kafka/source/GeodeSourceConnectorConfig.java
new file mode 100644
index 0000000..c29048f
--- /dev/null
+++ b/src/main/java/geode/kafka/source/GeodeSourceConnectorConfig.java
@@ -0,0 +1,104 @@
+package geode.kafka.source;
+
+import geode.kafka.GeodeConnectorConfig;
+import geode.kafka.LocatorHostPort;
+
+import java.util.List;
+import java.util.Map;
+
+public class GeodeSourceConnectorConfig extends GeodeConnectorConfig {
+
+ //Geode Configuration
+ public static final String DURABLE_CLIENT_ID_PREFIX = "durableClientId";
+ public static final String DEFAULT_DURABLE_CLIENT_ID = "";
+ public static final String DURABLE_CLIENT_TIME_OUT = "durableClientTimeout";
+ public static final String DEFAULT_DURABLE_CLIENT_TIMEOUT = "60000";
+
+ public static final String CQ_PREFIX = "cqPrefix";
+ public static final String DEFAULT_CQ_PREFIX = "cqForGeodeKafka";
+
+ /**
+ * Used as a key for source partitions
+ */
+ public static final String REGION = "region";
+
+ public static final String REGION_TO_TOPIC_BINDINGS = "regionToTopic";
+
+ public static final String BATCH_SIZE = "geodeConnectorBatchSize";
+ public static final String DEFAULT_BATCH_SIZE = "100";
+
+ public static final String QUEUE_SIZE = "geodeConnectorQueueSize";
+ public static final String DEFAULT_QUEUE_SIZE = "100000";
+
+ public static final String LOAD_ENTIRE_REGION = "loadEntireRegion";
+ public static final String DEFAULT_LOAD_ENTIRE_REGION = "false";
+
+ private final String durableClientId;
+ private final String durableClientIdPrefix;
+ private final String durableClientTimeout;
+ private final String cqPrefix;
+ private final boolean loadEntireRegion;
+
+ private Map<String, List<String>> regionToTopics;
+
+ //just for tests
+ protected GeodeSourceConnectorConfig() {
+ super();
+ durableClientId = "";
+ durableClientIdPrefix = "";
+ durableClientTimeout = "0";
+ cqPrefix = DEFAULT_CQ_PREFIX;
+ loadEntireRegion = Boolean.parseBoolean(DEFAULT_LOAD_ENTIRE_REGION);
+ }
+
+ public GeodeSourceConnectorConfig(Map<String, String> connectorProperties) {
+ super(connectorProperties);
+ regionToTopics = parseRegionToTopics(connectorProperties.get(REGION_TO_TOPIC_BINDINGS));
+ durableClientIdPrefix = connectorProperties.get(DURABLE_CLIENT_ID_PREFIX);
+ if (isDurable(durableClientIdPrefix)) {
+ durableClientId = durableClientIdPrefix + taskId;
+ } else {
+ durableClientId = "";
+ }
+ durableClientTimeout = connectorProperties.get(DURABLE_CLIENT_TIME_OUT);
+ cqPrefix = connectorProperties.get(CQ_PREFIX);
+ loadEntireRegion = Boolean.parseBoolean(connectorProperties.get(LOAD_ENTIRE_REGION));
+ }
+
+ public boolean isDurable() {
+ return isDurable(durableClientId);
+ }
+
+ /**
+ * @param durableClientId or prefix can be passed in. Either both will be "" or both will have a value
+ * @return
+ */
+ boolean isDurable(String durableClientId) {
+ return !durableClientId.equals("");
+ }
+
+ public int getTaskId() {
+ return taskId;
+ }
+
+ public String getDurableClientId() {
+ return durableClientId;
+ }
+
+ public String getDurableClientTimeout() {
+ return durableClientTimeout;
+ }
+
+ public String getCqPrefix() {
+ return cqPrefix;
+ }
+
+ public boolean getLoadEntireRegion() {
+ return loadEntireRegion;
+ }
+
+ public Map<String, List<String>> getRegionToTopics() {
+ return regionToTopics;
+ }
+
+}
diff --git a/src/test/java/geode/kafka/GeodeConnectorConfigTest.java b/src/test/java/geode/kafka/GeodeConnectorConfigTest.java
index 6a39c5d..904b981 100644
--- a/src/test/java/geode/kafka/GeodeConnectorConfigTest.java
+++ b/src/test/java/geode/kafka/GeodeConnectorConfigTest.java
@@ -10,8 +10,6 @@
import java.util.List;
import java.util.Map;
-import static geode.kafka.GeodeConnectorConfig.DURABLE_CLIENT_ID_PREFIX;
-import static geode.kafka.GeodeConnectorConfig.DURABLE_CLIENT_TIME_OUT;
import static geode.kafka.GeodeConnectorConfig.LOCATORS;
import static geode.kafka.GeodeConnectorConfig.TASK_ID;
import static org.hamcrest.CoreMatchers.allOf;
@@ -26,7 +24,7 @@
@Test
public void parseRegionNamesShouldSplitOnComma() {
GeodeConnectorConfig config = new GeodeConnectorConfig();
- List<String> regionNames = config.parseNames("region1,region2,region3,region4");
+ List<String> regionNames = config.parseStringByComma("region1,region2,region3,region4");
assertEquals(4, regionNames.size());
assertThat(true, allOf(is(regionNames.contains("region1"))
, is(regionNames.contains("region2"))
@@ -37,7 +35,7 @@
@Test
public void parseRegionNamesShouldChomp() {
GeodeConnectorConfig config = new GeodeConnectorConfig();
- List<String> regionNames = config.parseNames("region1, region2, region3,region4");
+ List<String> regionNames = config.parseStringByComma("region1, region2, region3,region4");
assertEquals(4, regionNames.size());
assertThat(true, allOf(is(regionNames instanceof List)
, is(regionNames.contains("region1"))
@@ -135,14 +133,5 @@
*/
- @Test
- public void durableClientIdShouldNotBeSetIfPrefixIsEmpty() {
- Map<String, String> props = new HashMap<>();
- props.put(TASK_ID, "0");
- props.put(DURABLE_CLIENT_ID_PREFIX, "");
- props.put(LOCATORS, "localhost[10334]");
- GeodeConnectorConfig config = new GeodeConnectorConfig(props);
- assertEquals("", config.getDurableClientId());
- }
}
diff --git a/src/test/java/geode/kafka/WorkerAndHerderWrapper.java b/src/test/java/geode/kafka/WorkerAndHerderWrapper.java
index 5e8f074..1b46fe0 100644
--- a/src/test/java/geode/kafka/WorkerAndHerderWrapper.java
+++ b/src/test/java/geode/kafka/WorkerAndHerderWrapper.java
@@ -18,12 +18,11 @@
import java.util.HashMap;
import java.util.Map;
-import static geode.kafka.GeodeConnectorConfig.REGION_TO_TOPIC_BINDINGS;
-import static geode.kafka.GeodeConnectorConfig.TOPICS;
-import static geode.kafka.GeodeConnectorConfig.TOPIC_TO_REGION_BINDINGS;
import static geode.kafka.GeodeKafkaTestCluster.TEST_REGION_TO_TOPIC_BINDINGS;
import static geode.kafka.GeodeKafkaTestCluster.TEST_TOPIC_FOR_SINK;
import static geode.kafka.GeodeKafkaTestCluster.TEST_TOPIC_TO_REGION_BINDINGS;
+import static geode.kafka.GeodeSinkConnectorConfig.TOPIC_TO_REGION_BINDINGS;
+import static geode.kafka.source.GeodeSourceConnectorConfig.REGION_TO_TOPIC_BINDINGS;
public class WorkerAndHerderWrapper {
@@ -34,9 +33,6 @@
// fast flushing for testing.
props.put(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG, "10");
-
- props.put(WorkerConfig.INTERNAL_KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
- props.put(WorkerConfig.INTERNAL_VALUE_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
props.put("internal.key.converter.schemas.enable", "false");
props.put("internal.value.converter.schemas.enable", "false");
props.put(WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
@@ -71,7 +67,7 @@
sinkProps.put(ConnectorConfig.NAME_CONFIG, "geode-kafka-sink-connector");
sinkProps.put(ConnectorConfig.TASKS_MAX_CONFIG, "1");
sinkProps.put(TOPIC_TO_REGION_BINDINGS, TEST_TOPIC_TO_REGION_BINDINGS);
- sinkProps.put(TOPICS, TEST_TOPIC_FOR_SINK);
+ sinkProps.put("topics", TEST_TOPIC_FOR_SINK);
herder.putConnectorConfig(
sinkProps.get(ConnectorConfig.NAME_CONFIG),
diff --git a/src/test/java/geode/kafka/sink/BatchRecordsTest.java b/src/test/java/geode/kafka/sink/BatchRecordsTest.java
index 4907d74..593fea0 100644
--- a/src/test/java/geode/kafka/sink/BatchRecordsTest.java
+++ b/src/test/java/geode/kafka/sink/BatchRecordsTest.java
@@ -1,37 +1,82 @@
package geode.kafka.sink;
+import org.apache.geode.cache.Region;
+import org.apache.kafka.connect.sink.SinkRecord;
import org.junit.Test;
+import java.util.Collection;
+import java.util.Map;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
public class BatchRecordsTest {
@Test
public void updatingARecordShouldRemoveFromTheRemoveListIfNullValuesIsRemoveBooleanIsSet() {
-
+ Map updates = mock(Map.class);
+ Collection removes = mock(Collection.class);
+ when(removes.contains(any())).thenReturn(true);
+ BatchRecords records = new BatchRecords(updates, removes);
+ SinkRecord sinkRecord = mock(SinkRecord.class);
+ records.addUpdateOperation(sinkRecord, true);
+ verify(removes, times(1)).remove(any());
}
@Test
public void updatingARecordShouldAddToTheUpdateMap() {
-
+ Map updates = mock(Map.class);
+ Collection removes = mock(Collection.class);
+ when(removes.contains(any())).thenReturn(false);
+ BatchRecords records = new BatchRecords(updates, removes);
+ SinkRecord sinkRecord = mock(SinkRecord.class);
+ records.addUpdateOperation(sinkRecord, true);
+ verify(updates, times(1)).put(any(), any());
}
@Test
public void updatingARecordShouldNotRemoveFromTheRemoveListIfNullValuesIsNotSet() {
-
+ boolean nullValuesMeanRemove = false;
+ Map updates = mock(Map.class);
+ Collection removes = mock(Collection.class);
+ when(removes.contains(any())).thenReturn(true);
+ BatchRecords records = new BatchRecords(updates, removes);
+ SinkRecord sinkRecord = mock(SinkRecord.class);
+ records.addUpdateOperation(sinkRecord, nullValuesMeanRemove);
+ verify(removes, times(0)).remove(any());
}
@Test
- public void removingARecordShouldRemoveFromTheUpdateMap() {
-
+ public void removingARecordShouldRemoveFromTheUpdateMapIfKeyIsPresent() {
+ Map updates = mock(Map.class);
+ Collection removes = mock(Collection.class);
+ when(updates.containsKey(any())).thenReturn(true);
+ BatchRecords records = new BatchRecords(updates, removes);
+ SinkRecord sinkRecord = mock(SinkRecord.class);
+ records.addRemoveOperation(sinkRecord);
+ verify(updates, times(1)).remove(any());
}
@Test
public void removingARecordAddToTheRemoveCollection() {
-
+ Map updates = mock(Map.class);
+ Collection removes = mock(Collection.class);
+ BatchRecords records = new BatchRecords(updates, removes);
+ SinkRecord sinkRecord = mock(SinkRecord.class);
+ records.addRemoveOperation(sinkRecord);
+ verify(removes, times(1)).add(any());
}
@Test
public void executeOperationsShouldInvokePutAllAndRemoveAll() {
-
+ Region region = mock(Region.class);
+ BatchRecords records = new BatchRecords();
+ records.executeOperations(region);
+ verify(region, times(1)).putAll(any());
+ verify(region, times(1)).removeAll(any());
}
diff --git a/src/test/java/geode/kafka/sink/GeodeKafkaSinkTaskTest.java b/src/test/java/geode/kafka/sink/GeodeKafkaSinkTaskTest.java
new file mode 100644
index 0000000..37131ba
--- /dev/null
+++ b/src/test/java/geode/kafka/sink/GeodeKafkaSinkTaskTest.java
@@ -0,0 +1,4 @@
+package geode.kafka.sink;
+
+public class GeodeKafkaSinkTaskTest {
+}
diff --git a/src/test/java/geode/kafka/sink/GeodeKafkaSinkTest.java b/src/test/java/geode/kafka/sink/GeodeKafkaSinkTest.java
new file mode 100644
index 0000000..2720e90
--- /dev/null
+++ b/src/test/java/geode/kafka/sink/GeodeKafkaSinkTest.java
@@ -0,0 +1,16 @@
+package geode.kafka.sink;
+
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class GeodeKafkaSinkTest {
+
+ @Test
+ public void test() {
+ GeodeKafkaSink sink = new GeodeKafkaSink();
+ Map<String, String> props = new HashMap();
+ sink.start(props);
+ }
+}
diff --git a/src/test/java/geode/kafka/source/GeodeKafkaSourceTaskTest.java b/src/test/java/geode/kafka/source/GeodeKafkaSourceTaskTest.java
index d4149db..b793d30 100644
--- a/src/test/java/geode/kafka/source/GeodeKafkaSourceTaskTest.java
+++ b/src/test/java/geode/kafka/source/GeodeKafkaSourceTaskTest.java
@@ -1,6 +1,5 @@
package geode.kafka.source;
-import geode.kafka.GeodeConnectorConfig;
import geode.kafka.GeodeContext;
import org.apache.geode.cache.client.ClientCache;
import org.apache.geode.cache.query.CqEvent;
@@ -15,8 +14,8 @@
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
-import static geode.kafka.GeodeConnectorConfig.DEFAULT_CQ_PREFIX;
-import static geode.kafka.GeodeConnectorConfig.REGION_NAME;
+import static geode.kafka.source.GeodeSourceConnectorConfig.DEFAULT_CQ_PREFIX;
+import static org.apache.geode.pdx.internal.PeerTypeRegistration.REGION_NAME;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
@@ -88,7 +87,7 @@
GeodeContext geodeContext = mock(GeodeContext.class);
when(geodeContext.getClientCache()).thenReturn(clientCache);
- GeodeConnectorConfig config = mock(GeodeConnectorConfig.class);
+ GeodeSourceConnectorConfig config = mock(GeodeSourceConnectorConfig.class);
when (config.isDurable()).thenReturn(true);
GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
task.installOnGeode(config, geodeContext, null, "", false);
@@ -105,7 +104,7 @@
Map<String, List<String>> regionToTopicsMap = new HashMap<>();
regionToTopicsMap.put("region1", new ArrayList());
- GeodeConnectorConfig config = mock(GeodeConnectorConfig.class);
+ GeodeSourceConnectorConfig config = mock(GeodeSourceConnectorConfig.class);
when (config.getRegionToTopics()).thenReturn(regionToTopicsMap);
GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
@@ -123,7 +122,7 @@
Map<String, List<String>> regionToTopicsMap = new HashMap<>();
regionToTopicsMap.put("region1", new ArrayList());
- GeodeConnectorConfig config = mock(GeodeConnectorConfig.class);
+ GeodeSourceConnectorConfig config = mock(GeodeSourceConnectorConfig.class);
when (config.getRegionToTopics()).thenReturn(regionToTopicsMap);
GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
@@ -138,7 +137,7 @@
GeodeContext geodeContext = mock(GeodeContext.class);
when(geodeContext.getClientCache()).thenReturn(clientCache);
- GeodeConnectorConfig config = mock(GeodeConnectorConfig.class);
+ GeodeSourceConnectorConfig config = mock(GeodeSourceConnectorConfig.class);
when (config.isDurable()).thenReturn(false);
GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
task.installOnGeode(config, geodeContext, null, "", false);
diff --git a/src/test/java/geode/kafka/source/GeodeSourceConnectorConfigTest.java b/src/test/java/geode/kafka/source/GeodeSourceConnectorConfigTest.java
new file mode 100644
index 0000000..629c07d
--- /dev/null
+++ b/src/test/java/geode/kafka/source/GeodeSourceConnectorConfigTest.java
@@ -0,0 +1,26 @@
+package geode.kafka.source;
+
+import geode.kafka.GeodeConnectorConfig;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static geode.kafka.GeodeConnectorConfig.LOCATORS;
+import static geode.kafka.GeodeConnectorConfig.TASK_ID;
+import static geode.kafka.source.GeodeSourceConnectorConfig.DURABLE_CLIENT_ID_PREFIX;
+import static org.junit.Assert.assertEquals;
+
+public class GeodeSourceConnectorConfigTest {
+
+ @Test
+ public void durableClientIdShouldNotBeSetIfPrefixIsEmpty() {
+ Map<String, String> props = new HashMap<>();
+ props.put(TASK_ID, "0");
+ props.put(DURABLE_CLIENT_ID_PREFIX, "");
+ props.put(LOCATORS, "localhost[10334]");
+ GeodeSourceConnectorConfig config = new GeodeSourceConnectorConfig(props);
+ assertEquals("", config.getDurableClientId());
+ }
+
+}
