Renaming the package name to org.apache.geode.kafka
diff --git a/src/main/java/geode/kafka/source/GeodeKafkaSource.java b/src/main/java/geode/kafka/source/GeodeKafkaSource.java
deleted file mode 100644
index f150e07..0000000
--- a/src/main/java/geode/kafka/source/GeodeKafkaSource.java
+++ /dev/null
@@ -1,113 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
- * agreements. See the NOTICE file distributed with this work for additional information regarding
- * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance with the License. You may obtain a
- * copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License
- * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
- * or implied. See the License for the specific language governing permissions and limitations under
- * the License.
- */
-package geode.kafka.source;
-
-import static geode.kafka.GeodeConnectorConfig.DEFAULT_LOCATOR;
-import static geode.kafka.GeodeConnectorConfig.LOCATORS;
-import static geode.kafka.source.GeodeSourceConnectorConfig.BATCH_SIZE;
-import static geode.kafka.source.GeodeSourceConnectorConfig.CQS_TO_REGISTER;
-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;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import geode.kafka.GeodeConnectorConfig;
-import org.apache.kafka.common.config.ConfigDef;
-import org.apache.kafka.common.utils.AppInfoParser;
-import org.apache.kafka.connect.connector.Task;
-import org.apache.kafka.connect.source.SourceConnector;
-import org.apache.kafka.connect.util.ConnectorUtils;
-
-
-public class GeodeKafkaSource extends SourceConnector {
-
- private Map<String, String> sharedProps;
- // TODO maybe club this into GeodeConnnectorConfig
- private static final ConfigDef CONFIG_DEF = new ConfigDef();
-
-
- @Override
- public Class<? extends Task> taskClass() {
- return GeodeKafkaSourceTask.class;
- }
-
- @Override
- public List<Map<String, String>> taskConfigs(int maxTasks) {
- List<Map<String, String>> taskConfigs = new ArrayList<>();
- List<String> bindings =
- GeodeConnectorConfig.parseStringByComma(sharedProps.get(REGION_TO_TOPIC_BINDINGS));
- List<List<String>> bindingsPerTask = ConnectorUtils.groupPartitions(bindings, maxTasks);
-
- for (int i = 0; i < maxTasks; i++) {
- Map<String, String> taskProps = new HashMap<>();
- taskProps.putAll(sharedProps);
- taskProps.put(GeodeConnectorConfig.TASK_ID, "" + i);
- taskProps.put(CQS_TO_REGISTER,
- GeodeConnectorConfig.reconstructString(bindingsPerTask.get(i)));
- taskConfigs.add(taskProps);
- }
- return taskConfigs;
- }
-
-
- @Override
- public ConfigDef config() {
- return CONFIG_DEF;
- }
-
- @Override
- public void start(Map<String, String> props) {
- sharedProps = computeMissingConfigurations(props);
- }
-
- 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(LOAD_ENTIRE_REGION, (key) -> DEFAULT_LOAD_ENTIRE_REGION);
- return props;
- }
-
- @Override
- public void stop() {
-
- }
-
- @Override
- public String version() {
- // TODO
- return AppInfoParser.getVersion();
- }
-
- public Map<String, String> getSharedProps() {
- return sharedProps;
- }
-}
diff --git a/src/main/java/geode/kafka/GeodeConnectorConfig.java b/src/main/java/org/geode/kafka/GeodeConnectorConfig.java
similarity index 99%
rename from src/main/java/geode/kafka/GeodeConnectorConfig.java
rename to src/main/java/org/geode/kafka/GeodeConnectorConfig.java
index 7702765..62ee160 100644
--- a/src/main/java/geode/kafka/GeodeConnectorConfig.java
+++ b/src/main/java/org/geode/kafka/GeodeConnectorConfig.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka;
+package org.geode.kafka;
import java.util.Arrays;
import java.util.Collection;
diff --git a/src/main/java/geode/kafka/GeodeContext.java b/src/main/java/org/geode/kafka/GeodeContext.java
similarity index 95%
rename from src/main/java/geode/kafka/GeodeContext.java
rename to src/main/java/org/geode/kafka/GeodeContext.java
index ba5f9e5..d844581 100644
--- a/src/main/java/geode/kafka/GeodeContext.java
+++ b/src/main/java/org/geode/kafka/GeodeContext.java
@@ -12,9 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka;
-
-import static geode.kafka.GeodeConnectorConfig.SECURITY_CLIENT_AUTH_INIT;
+package org.geode.kafka;
import java.util.Collection;
import java.util.List;
@@ -58,7 +56,7 @@
ClientCacheFactory ccf = new ClientCacheFactory();
if (securityAuthInit != null) {
- ccf.set(SECURITY_CLIENT_AUTH_INIT, securityAuthInit);
+ ccf.set(GeodeConnectorConfig.SECURITY_CLIENT_AUTH_INIT, securityAuthInit);
}
if (!durableClientName.equals("")) {
ccf.set("durable-client-id", durableClientName)
diff --git a/src/main/java/geode/kafka/LocatorHostPort.java b/src/main/java/org/geode/kafka/LocatorHostPort.java
similarity index 97%
rename from src/main/java/geode/kafka/LocatorHostPort.java
rename to src/main/java/org/geode/kafka/LocatorHostPort.java
index 191d42d..5c71fa1 100644
--- a/src/main/java/geode/kafka/LocatorHostPort.java
+++ b/src/main/java/org/geode/kafka/LocatorHostPort.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka;
+package org.geode.kafka;
public class LocatorHostPort {
diff --git a/src/main/java/geode/kafka/sink/BatchRecords.java b/src/main/java/org/geode/kafka/sink/BatchRecords.java
similarity index 98%
rename from src/main/java/geode/kafka/sink/BatchRecords.java
rename to src/main/java/org/geode/kafka/sink/BatchRecords.java
index 0221cbe..049abac 100644
--- a/src/main/java/geode/kafka/sink/BatchRecords.java
+++ b/src/main/java/org/geode/kafka/sink/BatchRecords.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka.sink;
+package org.geode.kafka.sink;
import java.util.ArrayList;
import java.util.Collection;
diff --git a/src/main/java/geode/kafka/sink/GeodeKafkaSink.java b/src/main/java/org/geode/kafka/sink/GeodeKafkaSink.java
similarity index 81%
rename from src/main/java/geode/kafka/sink/GeodeKafkaSink.java
rename to src/main/java/org/geode/kafka/sink/GeodeKafkaSink.java
index 550e8a9..bc768b1 100644
--- a/src/main/java/geode/kafka/sink/GeodeKafkaSink.java
+++ b/src/main/java/org/geode/kafka/sink/GeodeKafkaSink.java
@@ -12,19 +12,14 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka.sink;
-
-import static geode.kafka.GeodeConnectorConfig.DEFAULT_LOCATOR;
-import static geode.kafka.GeodeConnectorConfig.LOCATORS;
-import static geode.kafka.GeodeSinkConnectorConfig.DEFAULT_NULL_VALUES_MEAN_REMOVE;
-import static geode.kafka.GeodeSinkConnectorConfig.NULL_VALUES_MEAN_REMOVE;
+package org.geode.kafka.sink;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import geode.kafka.GeodeConnectorConfig;
+import org.geode.kafka.GeodeConnectorConfig;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.connect.connector.Task;
import org.apache.kafka.connect.sink.SinkConnector;
@@ -77,8 +72,10 @@
private Map<String, String> computeMissingConfigurations(Map<String, String> props) {
- props.computeIfAbsent(LOCATORS, (key) -> DEFAULT_LOCATOR);
- props.computeIfAbsent(NULL_VALUES_MEAN_REMOVE, (key) -> DEFAULT_NULL_VALUES_MEAN_REMOVE);
+ props.computeIfAbsent(
+ GeodeConnectorConfig.LOCATORS, (key) -> GeodeConnectorConfig.DEFAULT_LOCATOR);
+ props.computeIfAbsent(
+ GeodeSinkConnectorConfig.NULL_VALUES_MEAN_REMOVE, (key) -> GeodeSinkConnectorConfig.DEFAULT_NULL_VALUES_MEAN_REMOVE);
return props;
}
}
diff --git a/src/main/java/geode/kafka/sink/GeodeKafkaSinkTask.java b/src/main/java/org/geode/kafka/sink/GeodeKafkaSinkTask.java
similarity index 97%
rename from src/main/java/geode/kafka/sink/GeodeKafkaSinkTask.java
rename to src/main/java/org/geode/kafka/sink/GeodeKafkaSinkTask.java
index 3552b2d..b019d80 100644
--- a/src/main/java/geode/kafka/sink/GeodeKafkaSinkTask.java
+++ b/src/main/java/org/geode/kafka/sink/GeodeKafkaSinkTask.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka.sink;
+package org.geode.kafka.sink;
import java.util.Collection;
import java.util.HashMap;
@@ -20,8 +20,7 @@
import java.util.Map;
import java.util.stream.Collectors;
-import geode.kafka.GeodeContext;
-import geode.kafka.GeodeSinkConnectorConfig;
+import org.geode.kafka.GeodeContext;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.sink.SinkTask;
import org.slf4j.Logger;
diff --git a/src/main/java/geode/kafka/sink/GeodeSinkConnectorConfig.java b/src/main/java/org/geode/kafka/sink/GeodeSinkConnectorConfig.java
similarity index 95%
rename from src/main/java/geode/kafka/sink/GeodeSinkConnectorConfig.java
rename to src/main/java/org/geode/kafka/sink/GeodeSinkConnectorConfig.java
index e416a9f..558b874 100644
--- a/src/main/java/geode/kafka/sink/GeodeSinkConnectorConfig.java
+++ b/src/main/java/org/geode/kafka/sink/GeodeSinkConnectorConfig.java
@@ -12,11 +12,13 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka;
+package org.geode.kafka.sink;
import java.util.List;
import java.util.Map;
+import org.geode.kafka.GeodeConnectorConfig;
+
public class GeodeSinkConnectorConfig extends GeodeConnectorConfig {
// Used by sink
public static final String TOPIC_TO_REGION_BINDINGS = "topicToRegions";
diff --git a/src/main/java/geode/kafka/source/EventBufferSupplier.java b/src/main/java/org/geode/kafka/source/EventBufferSupplier.java
similarity index 96%
rename from src/main/java/geode/kafka/source/EventBufferSupplier.java
rename to src/main/java/org/geode/kafka/source/EventBufferSupplier.java
index 8b05e22..be40602 100644
--- a/src/main/java/geode/kafka/source/EventBufferSupplier.java
+++ b/src/main/java/org/geode/kafka/source/EventBufferSupplier.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka.source;
+package org.geode.kafka.source;
import java.util.concurrent.BlockingQueue;
import java.util.function.Supplier;
diff --git a/src/main/java/geode/kafka/source/GeodeEvent.java b/src/main/java/org/geode/kafka/source/GeodeEvent.java
similarity index 97%
rename from src/main/java/geode/kafka/source/GeodeEvent.java
rename to src/main/java/org/geode/kafka/source/GeodeEvent.java
index f569f2b..2333955 100644
--- a/src/main/java/geode/kafka/source/GeodeEvent.java
+++ b/src/main/java/org/geode/kafka/source/GeodeEvent.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka.source;
+package org.geode.kafka.source;
import org.apache.geode.cache.query.CqEvent;
diff --git a/src/main/java/org/geode/kafka/source/GeodeKafkaSource.java b/src/main/java/org/geode/kafka/source/GeodeKafkaSource.java
new file mode 100644
index 0000000..be4e7a5
--- /dev/null
+++ b/src/main/java/org/geode/kafka/source/GeodeKafkaSource.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.geode.kafka.source;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.geode.kafka.GeodeConnectorConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.utils.AppInfoParser;
+import org.apache.kafka.connect.connector.Task;
+import org.apache.kafka.connect.source.SourceConnector;
+import org.apache.kafka.connect.util.ConnectorUtils;
+
+
+public class GeodeKafkaSource extends SourceConnector {
+
+ private Map<String, String> sharedProps;
+ // TODO maybe club this into GeodeConnnectorConfig
+ private static final ConfigDef CONFIG_DEF = new ConfigDef();
+
+
+ @Override
+ public Class<? extends Task> taskClass() {
+ return GeodeKafkaSourceTask.class;
+ }
+
+ @Override
+ public List<Map<String, String>> taskConfigs(int maxTasks) {
+ List<Map<String, String>> taskConfigs = new ArrayList<>();
+ List<String> bindings =
+ GeodeConnectorConfig
+ .parseStringByComma(sharedProps.get(GeodeSourceConnectorConfig.REGION_TO_TOPIC_BINDINGS));
+ List<List<String>> bindingsPerTask = ConnectorUtils.groupPartitions(bindings, maxTasks);
+
+ for (int i = 0; i < maxTasks; i++) {
+ Map<String, String> taskProps = new HashMap<>();
+ taskProps.putAll(sharedProps);
+ taskProps.put(GeodeConnectorConfig.TASK_ID, "" + i);
+ taskProps.put(GeodeSourceConnectorConfig.CQS_TO_REGISTER,
+ GeodeConnectorConfig.reconstructString(bindingsPerTask.get(i)));
+ taskConfigs.add(taskProps);
+ }
+ return taskConfigs;
+ }
+
+
+ @Override
+ public ConfigDef config() {
+ return CONFIG_DEF;
+ }
+
+ @Override
+ public void start(Map<String, String> props) {
+ sharedProps = computeMissingConfigurations(props);
+ }
+
+ private Map<String, String> computeMissingConfigurations(Map<String, String> props) {
+ props.computeIfAbsent(
+ GeodeConnectorConfig.LOCATORS, (key) -> GeodeConnectorConfig.DEFAULT_LOCATOR);
+ props.computeIfAbsent(
+ GeodeSourceConnectorConfig.DURABLE_CLIENT_TIME_OUT, (key) -> GeodeSourceConnectorConfig.DEFAULT_DURABLE_CLIENT_TIMEOUT);
+ props.computeIfAbsent(
+ GeodeSourceConnectorConfig.DURABLE_CLIENT_ID_PREFIX, (key) -> GeodeSourceConnectorConfig.DEFAULT_DURABLE_CLIENT_ID);
+ props.computeIfAbsent(
+ GeodeSourceConnectorConfig.BATCH_SIZE, (key) -> GeodeSourceConnectorConfig.DEFAULT_BATCH_SIZE);
+ props.computeIfAbsent(
+ GeodeSourceConnectorConfig.QUEUE_SIZE, (key) -> GeodeSourceConnectorConfig.DEFAULT_QUEUE_SIZE);
+ props.computeIfAbsent(
+ GeodeSourceConnectorConfig.CQ_PREFIX, (key) -> GeodeSourceConnectorConfig.DEFAULT_CQ_PREFIX);
+ props.computeIfAbsent(
+ GeodeSourceConnectorConfig.LOAD_ENTIRE_REGION, (key) -> GeodeSourceConnectorConfig.DEFAULT_LOAD_ENTIRE_REGION);
+ return props;
+ }
+
+ @Override
+ public void stop() {
+
+ }
+
+ @Override
+ public String version() {
+ // TODO
+ return AppInfoParser.getVersion();
+ }
+
+ public Map<String, String> getSharedProps() {
+ return sharedProps;
+ }
+}
diff --git a/src/main/java/geode/kafka/source/GeodeKafkaSourceListener.java b/src/main/java/org/geode/kafka/source/GeodeKafkaSourceListener.java
similarity index 98%
rename from src/main/java/geode/kafka/source/GeodeKafkaSourceListener.java
rename to src/main/java/org/geode/kafka/source/GeodeKafkaSourceListener.java
index b0b8c6a..e875ee4 100644
--- a/src/main/java/geode/kafka/source/GeodeKafkaSourceListener.java
+++ b/src/main/java/org/geode/kafka/source/GeodeKafkaSourceListener.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka.source;
+package org.geode.kafka.source;
import java.util.concurrent.TimeUnit;
diff --git a/src/main/java/geode/kafka/source/GeodeKafkaSourceTask.java b/src/main/java/org/geode/kafka/source/GeodeKafkaSourceTask.java
similarity index 92%
rename from src/main/java/geode/kafka/source/GeodeKafkaSourceTask.java
rename to src/main/java/org/geode/kafka/source/GeodeKafkaSourceTask.java
index da2119a..2b08973 100644
--- a/src/main/java/geode/kafka/source/GeodeKafkaSourceTask.java
+++ b/src/main/java/org/geode/kafka/source/GeodeKafkaSourceTask.java
@@ -12,11 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka.source;
-
-import static geode.kafka.source.GeodeSourceConnectorConfig.BATCH_SIZE;
-import static geode.kafka.source.GeodeSourceConnectorConfig.QUEUE_SIZE;
-import static geode.kafka.source.GeodeSourceConnectorConfig.REGION_PARTITION;
+package org.geode.kafka.source;
import java.util.ArrayList;
import java.util.Collection;
@@ -25,7 +21,7 @@
import java.util.Map;
import java.util.stream.Collectors;
-import geode.kafka.GeodeContext;
+import org.geode.kafka.GeodeContext;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTask;
import org.slf4j.Logger;
@@ -75,8 +71,9 @@
geodeConnectorConfig.getDurableClientId(), geodeConnectorConfig.getDurableClientTimeout(),
geodeConnectorConfig.getSecurityClientAuthInit());
- batchSize = Integer.parseInt(props.get(BATCH_SIZE));
- eventBufferSupplier = new SharedEventBufferSupplier(Integer.parseInt(props.get(QUEUE_SIZE)));
+ batchSize = Integer.parseInt(props.get(GeodeSourceConnectorConfig.BATCH_SIZE));
+ eventBufferSupplier = new SharedEventBufferSupplier(Integer.parseInt(props.get(
+ GeodeSourceConnectorConfig.QUEUE_SIZE)));
regionToTopics = geodeConnectorConfig.getRegionToTopics();
geodeConnectorConfig.getCqsToRegister();
@@ -165,9 +162,9 @@
Map<String, Map<String, String>> createSourcePartitionsMap(Collection<String> regionNames) {
return regionNames.stream().map(regionName -> {
Map<String, String> sourcePartition = new HashMap<>();
- sourcePartition.put(REGION_PARTITION, regionName);
+ sourcePartition.put(GeodeSourceConnectorConfig.REGION_PARTITION, regionName);
return sourcePartition;
- }).collect(Collectors.toMap(s -> s.get(REGION_PARTITION), s -> s));
+ }).collect(Collectors.toMap(s -> s.get(GeodeSourceConnectorConfig.REGION_PARTITION), s -> s));
}
String generateCqName(int taskId, String cqPrefix, String regionName) {
diff --git a/src/main/java/geode/kafka/source/GeodeSourceConnectorConfig.java b/src/main/java/org/geode/kafka/source/GeodeSourceConnectorConfig.java
similarity index 97%
rename from src/main/java/geode/kafka/source/GeodeSourceConnectorConfig.java
rename to src/main/java/org/geode/kafka/source/GeodeSourceConnectorConfig.java
index 4f0393d..82bb712 100644
--- a/src/main/java/geode/kafka/source/GeodeSourceConnectorConfig.java
+++ b/src/main/java/org/geode/kafka/source/GeodeSourceConnectorConfig.java
@@ -12,13 +12,13 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka.source;
+package org.geode.kafka.source;
import java.util.Collection;
import java.util.List;
import java.util.Map;
-import geode.kafka.GeodeConnectorConfig;
+import org.geode.kafka.GeodeConnectorConfig;
public class GeodeSourceConnectorConfig extends GeodeConnectorConfig {
diff --git a/src/main/java/geode/kafka/source/SharedEventBufferSupplier.java b/src/main/java/org/geode/kafka/source/SharedEventBufferSupplier.java
similarity index 98%
rename from src/main/java/geode/kafka/source/SharedEventBufferSupplier.java
rename to src/main/java/org/geode/kafka/source/SharedEventBufferSupplier.java
index 6ac6bb6..963a132 100644
--- a/src/main/java/geode/kafka/source/SharedEventBufferSupplier.java
+++ b/src/main/java/org/geode/kafka/source/SharedEventBufferSupplier.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka.source;
+package org.geode.kafka.source;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
diff --git a/src/test/java/geode/kafka/GeodeConnectorConfigTest.java b/src/test/java/org/geode/kafka/GeodeConnectorConfigTest.java
similarity index 99%
rename from src/test/java/geode/kafka/GeodeConnectorConfigTest.java
rename to src/test/java/org/geode/kafka/GeodeConnectorConfigTest.java
index c6fe491..e04db5a 100644
--- a/src/test/java/geode/kafka/GeodeConnectorConfigTest.java
+++ b/src/test/java/org/geode/kafka/GeodeConnectorConfigTest.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka;
+package org.geode.kafka;
import static org.hamcrest.CoreMatchers.allOf;
import static org.hamcrest.CoreMatchers.is;
diff --git a/src/test/java/geode/kafka/GeodeContextTest.java b/src/test/java/org/geode/kafka/GeodeContextTest.java
similarity index 96%
rename from src/test/java/geode/kafka/GeodeContextTest.java
rename to src/test/java/org/geode/kafka/GeodeContextTest.java
index 1f92bbb..eb10bee 100644
--- a/src/test/java/geode/kafka/GeodeContextTest.java
+++ b/src/test/java/org/geode/kafka/GeodeContextTest.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka;
+package org.geode.kafka;
public class GeodeContextTest {
}
diff --git a/src/test/java/geode/kafka/GeodeKafkaTestCluster.java b/src/test/java/org/geode/kafka/GeodeKafkaTestCluster.java
similarity index 99%
rename from src/test/java/geode/kafka/GeodeKafkaTestCluster.java
rename to src/test/java/org/geode/kafka/GeodeKafkaTestCluster.java
index c5acc3d..f620caa 100644
--- a/src/test/java/geode/kafka/GeodeKafkaTestCluster.java
+++ b/src/test/java/org/geode/kafka/GeodeKafkaTestCluster.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka;
+package org.geode.kafka;
import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
diff --git a/src/test/java/geode/kafka/GeodeLocalCluster.java b/src/test/java/org/geode/kafka/GeodeLocalCluster.java
similarity index 97%
rename from src/test/java/geode/kafka/GeodeLocalCluster.java
rename to src/test/java/org/geode/kafka/GeodeLocalCluster.java
index 259d30b..6784391 100644
--- a/src/test/java/geode/kafka/GeodeLocalCluster.java
+++ b/src/test/java/org/geode/kafka/GeodeLocalCluster.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka;
+package org.geode.kafka;
import java.io.IOException;
diff --git a/src/test/java/geode/kafka/JavaProcess.java b/src/test/java/org/geode/kafka/JavaProcess.java
similarity index 98%
rename from src/test/java/geode/kafka/JavaProcess.java
rename to src/test/java/org/geode/kafka/JavaProcess.java
index b130223..5355c75 100644
--- a/src/test/java/geode/kafka/JavaProcess.java
+++ b/src/test/java/org/geode/kafka/JavaProcess.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka;
+package org.geode.kafka;
import java.io.File;
import java.io.IOException;
diff --git a/src/test/java/geode/kafka/KafkaLocalCluster.java b/src/test/java/org/geode/kafka/KafkaLocalCluster.java
similarity index 98%
rename from src/test/java/geode/kafka/KafkaLocalCluster.java
rename to src/test/java/org/geode/kafka/KafkaLocalCluster.java
index ef534d9..ee13f8c 100644
--- a/src/test/java/geode/kafka/KafkaLocalCluster.java
+++ b/src/test/java/org/geode/kafka/KafkaLocalCluster.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka;
+package org.geode.kafka;
import java.io.IOException;
import java.util.Properties;
diff --git a/src/test/java/geode/kafka/LocatorLauncherWrapper.java b/src/test/java/org/geode/kafka/LocatorLauncherWrapper.java
similarity index 98%
rename from src/test/java/geode/kafka/LocatorLauncherWrapper.java
rename to src/test/java/org/geode/kafka/LocatorLauncherWrapper.java
index 1b922c9..bda962e 100644
--- a/src/test/java/geode/kafka/LocatorLauncherWrapper.java
+++ b/src/test/java/org/geode/kafka/LocatorLauncherWrapper.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka;
+package org.geode.kafka;
import java.io.File;
import java.io.IOException;
diff --git a/src/test/java/geode/kafka/ServerLauncherWrapper.java b/src/test/java/org/geode/kafka/ServerLauncherWrapper.java
similarity index 88%
rename from src/test/java/geode/kafka/ServerLauncherWrapper.java
rename to src/test/java/org/geode/kafka/ServerLauncherWrapper.java
index f60af27..1bc63fe 100644
--- a/src/test/java/geode/kafka/ServerLauncherWrapper.java
+++ b/src/test/java/org/geode/kafka/ServerLauncherWrapper.java
@@ -12,10 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka;
-
-import static geode.kafka.GeodeKafkaTestCluster.TEST_REGION_FOR_SINK;
-import static geode.kafka.GeodeKafkaTestCluster.TEST_REGION_FOR_SOURCE;
+package org.geode.kafka;
import java.io.IOException;
import java.util.Properties;
@@ -61,8 +58,10 @@
cacheServer.start();
// create the region
- cache.createRegionFactory(RegionShortcut.PARTITION).create(TEST_REGION_FOR_SINK);
- cache.createRegionFactory(RegionShortcut.PARTITION).create(TEST_REGION_FOR_SOURCE);
+ cache.createRegionFactory(RegionShortcut.PARTITION).create(
+ GeodeKafkaTestCluster.TEST_REGION_FOR_SINK);
+ cache.createRegionFactory(RegionShortcut.PARTITION).create(
+ GeodeKafkaTestCluster.TEST_REGION_FOR_SOURCE);
System.out.println("starting cacheserver");
while (true) {
diff --git a/src/test/java/geode/kafka/WorkerAndHerderCluster.java b/src/test/java/org/geode/kafka/WorkerAndHerderCluster.java
similarity index 97%
rename from src/test/java/geode/kafka/WorkerAndHerderCluster.java
rename to src/test/java/org/geode/kafka/WorkerAndHerderCluster.java
index 70461a3..824edab 100644
--- a/src/test/java/geode/kafka/WorkerAndHerderCluster.java
+++ b/src/test/java/org/geode/kafka/WorkerAndHerderCluster.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka;
+package org.geode.kafka;
import java.io.IOException;
diff --git a/src/test/java/geode/kafka/WorkerAndHerderWrapper.java b/src/test/java/org/geode/kafka/WorkerAndHerderWrapper.java
similarity index 84%
rename from src/test/java/geode/kafka/WorkerAndHerderWrapper.java
rename to src/test/java/org/geode/kafka/WorkerAndHerderWrapper.java
index 5f86985..b4a7bbe 100644
--- a/src/test/java/geode/kafka/WorkerAndHerderWrapper.java
+++ b/src/test/java/org/geode/kafka/WorkerAndHerderWrapper.java
@@ -12,20 +12,17 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka;
+package org.geode.kafka;
-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;
+import static org.geode.kafka.sink.GeodeSinkConnectorConfig.TOPIC_TO_REGION_BINDINGS;
+import static org.geode.kafka.source.GeodeSourceConnectorConfig.REGION_TO_TOPIC_BINDINGS;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
-import geode.kafka.sink.GeodeKafkaSink;
-import geode.kafka.source.GeodeKafkaSource;
+import org.geode.kafka.sink.GeodeKafkaSink;
+import org.geode.kafka.source.GeodeKafkaSource;
import org.apache.kafka.common.utils.SystemTime;
import org.apache.kafka.connect.connector.policy.AllConnectorClientConfigOverridePolicy;
import org.apache.kafka.connect.runtime.ConnectorConfig;
@@ -75,7 +72,7 @@
sourceProps.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, GeodeKafkaSource.class.getName());
sourceProps.put(ConnectorConfig.NAME_CONFIG, "geode-kafka-source-connector");
sourceProps.put(ConnectorConfig.TASKS_MAX_CONFIG, maxTasks);
- sourceProps.put(REGION_TO_TOPIC_BINDINGS, TEST_REGION_TO_TOPIC_BINDINGS);
+ sourceProps.put(REGION_TO_TOPIC_BINDINGS, GeodeKafkaTestCluster.TEST_REGION_TO_TOPIC_BINDINGS);
herder.putConnectorConfig(
sourceProps.get(ConnectorConfig.NAME_CONFIG),
@@ -86,8 +83,8 @@
sinkProps.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, GeodeKafkaSink.class.getName());
sinkProps.put(ConnectorConfig.NAME_CONFIG, "geode-kafka-sink-connector");
sinkProps.put(ConnectorConfig.TASKS_MAX_CONFIG, maxTasks);
- sinkProps.put(TOPIC_TO_REGION_BINDINGS, TEST_TOPIC_TO_REGION_BINDINGS);
- sinkProps.put("topics", TEST_TOPIC_FOR_SINK);
+ sinkProps.put(TOPIC_TO_REGION_BINDINGS, GeodeKafkaTestCluster.TEST_TOPIC_TO_REGION_BINDINGS);
+ sinkProps.put("topics", GeodeKafkaTestCluster.TEST_TOPIC_FOR_SINK);
herder.putConnectorConfig(
sinkProps.get(ConnectorConfig.NAME_CONFIG),
diff --git a/src/test/java/geode/kafka/ZooKeeperLocalCluster.java b/src/test/java/org/geode/kafka/ZooKeeperLocalCluster.java
similarity index 98%
rename from src/test/java/geode/kafka/ZooKeeperLocalCluster.java
rename to src/test/java/org/geode/kafka/ZooKeeperLocalCluster.java
index 717b046..65ff819 100644
--- a/src/test/java/geode/kafka/ZooKeeperLocalCluster.java
+++ b/src/test/java/org/geode/kafka/ZooKeeperLocalCluster.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka;
+package org.geode.kafka;
import java.io.IOException;
import java.util.Properties;
diff --git a/src/test/java/geode/kafka/sink/BatchRecordsTest.java b/src/test/java/org/geode/kafka/sink/BatchRecordsTest.java
similarity index 98%
rename from src/test/java/geode/kafka/sink/BatchRecordsTest.java
rename to src/test/java/org/geode/kafka/sink/BatchRecordsTest.java
index cdb286b..c2da554 100644
--- a/src/test/java/geode/kafka/sink/BatchRecordsTest.java
+++ b/src/test/java/org/geode/kafka/sink/BatchRecordsTest.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka.sink;
+package org.geode.kafka.sink;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
diff --git a/src/test/java/geode/kafka/sink/GeodeKafkaSinkTaskTest.java b/src/test/java/org/geode/kafka/sink/GeodeKafkaSinkTaskTest.java
similarity index 86%
rename from src/test/java/geode/kafka/sink/GeodeKafkaSinkTaskTest.java
rename to src/test/java/org/geode/kafka/sink/GeodeKafkaSinkTaskTest.java
index 735f255..f7feb6f 100644
--- a/src/test/java/geode/kafka/sink/GeodeKafkaSinkTaskTest.java
+++ b/src/test/java/org/geode/kafka/sink/GeodeKafkaSinkTaskTest.java
@@ -12,12 +12,10 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka.sink;
+package org.geode.kafka.sink;
-import static geode.kafka.GeodeConnectorConfig.LOCATORS;
-import static geode.kafka.GeodeConnectorConfig.TASK_ID;
-import static geode.kafka.GeodeSinkConnectorConfig.NULL_VALUES_MEAN_REMOVE;
-import static geode.kafka.GeodeSinkConnectorConfig.TOPIC_TO_REGION_BINDINGS;
+import static org.geode.kafka.sink.GeodeSinkConnectorConfig.NULL_VALUES_MEAN_REMOVE;
+import static org.geode.kafka.sink.GeodeSinkConnectorConfig.TOPIC_TO_REGION_BINDINGS;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
@@ -28,8 +26,8 @@
import java.util.HashMap;
import java.util.List;
-import geode.kafka.GeodeSinkConnectorConfig;
import org.apache.kafka.connect.sink.SinkRecord;
+import org.geode.kafka.GeodeConnectorConfig;
import org.junit.Test;
import org.apache.geode.cache.Region;
@@ -39,9 +37,9 @@
private HashMap<String, String> createTestSinkProps(boolean nullMeansRemove) {
HashMap<String, String> props = new HashMap<>();
props.put(TOPIC_TO_REGION_BINDINGS, "[topic:region]");
- props.put(TASK_ID, "0");
+ props.put(GeodeConnectorConfig.TASK_ID, "0");
props.put(NULL_VALUES_MEAN_REMOVE, String.valueOf(nullMeansRemove));
- props.put(LOCATORS, "localhost[10334]");
+ props.put(GeodeConnectorConfig.LOCATORS, "localhost[10334]");
return props;
}
diff --git a/src/test/java/geode/kafka/sink/GeodeKafkaSinkTest.java b/src/test/java/org/geode/kafka/sink/GeodeKafkaSinkTest.java
similarity index 90%
rename from src/test/java/geode/kafka/sink/GeodeKafkaSinkTest.java
rename to src/test/java/org/geode/kafka/sink/GeodeKafkaSinkTest.java
index d45c0c8..c82fec4 100644
--- a/src/test/java/geode/kafka/sink/GeodeKafkaSinkTest.java
+++ b/src/test/java/org/geode/kafka/sink/GeodeKafkaSinkTest.java
@@ -12,10 +12,9 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka.sink;
+package org.geode.kafka.sink;
-import static geode.kafka.GeodeConnectorConfig.TASK_ID;
-import static geode.kafka.GeodeSinkConnectorConfig.TOPIC_TO_REGION_BINDINGS;
+import static org.geode.kafka.sink.GeodeSinkConnectorConfig.TOPIC_TO_REGION_BINDINGS;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -24,6 +23,7 @@
import java.util.HashSet;
import java.util.Map;
+import org.geode.kafka.GeodeConnectorConfig;
import org.junit.Test;
public class GeodeKafkaSinkTest {
@@ -59,7 +59,7 @@
Collection<Map<String, String>> tasks = sink.taskConfigs(5);
HashSet<String> seenIds = new HashSet();
for (Map<String, String> taskProp : tasks) {
- assertTrue(seenIds.add(taskProp.get(TASK_ID)));
+ assertTrue(seenIds.add(taskProp.get(GeodeConnectorConfig.TASK_ID)));
}
}
}
diff --git a/src/test/java/geode/kafka/source/GeodeKafkaSourceTaskTest.java b/src/test/java/org/geode/kafka/source/GeodeKafkaSourceTaskTest.java
similarity index 97%
rename from src/test/java/geode/kafka/source/GeodeKafkaSourceTaskTest.java
rename to src/test/java/org/geode/kafka/source/GeodeKafkaSourceTaskTest.java
index a919b96..29a901a 100644
--- a/src/test/java/geode/kafka/source/GeodeKafkaSourceTaskTest.java
+++ b/src/test/java/org/geode/kafka/source/GeodeKafkaSourceTaskTest.java
@@ -12,10 +12,10 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka.source;
+package org.geode.kafka.source;
-import static geode.kafka.source.GeodeSourceConnectorConfig.DEFAULT_CQ_PREFIX;
-import static geode.kafka.source.GeodeSourceConnectorConfig.REGION_PARTITION;
+import static org.geode.kafka.source.GeodeSourceConnectorConfig.DEFAULT_CQ_PREFIX;
+import static org.geode.kafka.source.GeodeSourceConnectorConfig.REGION_PARTITION;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
@@ -36,7 +36,7 @@
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
-import geode.kafka.GeodeContext;
+import org.geode.kafka.GeodeContext;
import org.junit.Test;
import org.apache.geode.cache.client.ClientCache;
diff --git a/src/test/java/geode/kafka/source/GeodeKafkaSourceTest.java b/src/test/java/org/geode/kafka/source/GeodeKafkaSourceTest.java
similarity index 90%
rename from src/test/java/geode/kafka/source/GeodeKafkaSourceTest.java
rename to src/test/java/org/geode/kafka/source/GeodeKafkaSourceTest.java
index 6632d75..433550a 100644
--- a/src/test/java/geode/kafka/source/GeodeKafkaSourceTest.java
+++ b/src/test/java/org/geode/kafka/source/GeodeKafkaSourceTest.java
@@ -12,10 +12,9 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka.source;
+package org.geode.kafka.source;
-import static geode.kafka.GeodeConnectorConfig.TASK_ID;
-import static geode.kafka.source.GeodeSourceConnectorConfig.REGION_TO_TOPIC_BINDINGS;
+import static org.geode.kafka.source.GeodeSourceConnectorConfig.REGION_TO_TOPIC_BINDINGS;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -24,6 +23,7 @@
import java.util.HashSet;
import java.util.Map;
+import org.geode.kafka.GeodeConnectorConfig;
import org.junit.Test;
public class GeodeKafkaSourceTest {
@@ -59,7 +59,7 @@
Collection<Map<String, String>> tasks = sink.taskConfigs(5);
HashSet<String> seenIds = new HashSet();
for (Map<String, String> taskProp : tasks) {
- assertTrue(seenIds.add(taskProp.get(TASK_ID)));
+ assertTrue(seenIds.add(taskProp.get(GeodeConnectorConfig.TASK_ID)));
}
}
diff --git a/src/test/java/geode/kafka/source/GeodeSourceConnectorConfigTest.java b/src/test/java/org/geode/kafka/source/GeodeSourceConnectorConfigTest.java
similarity index 80%
rename from src/test/java/geode/kafka/source/GeodeSourceConnectorConfigTest.java
rename to src/test/java/org/geode/kafka/source/GeodeSourceConnectorConfigTest.java
index a545a72..fdcd7d3 100644
--- a/src/test/java/geode/kafka/source/GeodeSourceConnectorConfigTest.java
+++ b/src/test/java/org/geode/kafka/source/GeodeSourceConnectorConfigTest.java
@@ -12,16 +12,15 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka.source;
+package org.geode.kafka.source;
-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.geode.kafka.source.GeodeSourceConnectorConfig.DURABLE_CLIENT_ID_PREFIX;
import static org.junit.Assert.assertEquals;
import java.util.HashMap;
import java.util.Map;
+import org.geode.kafka.GeodeConnectorConfig;
import org.junit.Test;
public class GeodeSourceConnectorConfigTest {
@@ -29,9 +28,9 @@
@Test
public void durableClientIdShouldNotBeSetIfPrefixIsEmpty() {
Map<String, String> props = new HashMap<>();
- props.put(TASK_ID, "0");
+ props.put(GeodeConnectorConfig.TASK_ID, "0");
props.put(DURABLE_CLIENT_ID_PREFIX, "");
- props.put(LOCATORS, "localhost[10334]");
+ props.put(GeodeConnectorConfig.LOCATORS, "localhost[10334]");
GeodeSourceConnectorConfig config = new GeodeSourceConnectorConfig(props);
assertEquals("", config.getDurableClientId());
}
diff --git a/src/test/java/geode/kafka/source/SharedEventBufferSupplierTest.java b/src/test/java/org/geode/kafka/source/SharedEventBufferSupplierTest.java
similarity index 98%
rename from src/test/java/geode/kafka/source/SharedEventBufferSupplierTest.java
rename to src/test/java/org/geode/kafka/source/SharedEventBufferSupplierTest.java
index 87a0e07..92de30d 100644
--- a/src/test/java/geode/kafka/source/SharedEventBufferSupplierTest.java
+++ b/src/test/java/org/geode/kafka/source/SharedEventBufferSupplierTest.java
@@ -12,7 +12,7 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package geode.kafka.source;
+package org.geode.kafka.source;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
