Tests: Added offset resetting to the embedded Kafka

Author: rozza

src/integrationTest/java/com/mongodb/kafka/connect/embedded/ConnectStandalone.java

@@ -37,7 +37,7 @@
 import org.apache.kafka.connect.runtime.rest.entities.ConnectorInfo;
 import org.apache.kafka.connect.runtime.standalone.StandaloneConfig;
 import org.apache.kafka.connect.runtime.standalone.StandaloneHerder;
-import org.apache.kafka.connect.storage.FileOffsetBackingStore;
+import org.apache.kafka.connect.storage.MemoryOffsetBackingStore;
 import org.apache.kafka.connect.util.ConnectUtils;
 import org.apache.kafka.connect.util.FutureCallback;
 import org.slf4j.Logger;
@@ -51,6 +51,8 @@ class ConnectStandalone {
   private final Herder herder;
   private final Connect connect;
 
+  private final ResettableOffsetStore resettableOffsetStore;
+
   @SuppressWarnings("unchecked")
   ConnectStandalone(final Properties workerProperties) {
     Time time = Time.SYSTEM;
@@ -60,6 +62,7 @@ class ConnectStandalone {
     initInfo.logAll();
 
     Map<String, String> workerProps = (Map) workerProperties;
+    resettableOffsetStore = new ResettableOffsetStore();
 
     LOGGER.info("Scanning for plugin classes. This might take a moment ...");
     Plugins plugins = new Plugins(workerProps);
@@ -78,12 +81,7 @@ class ConnectStandalone {
         new NoneConnectorClientConfigOverridePolicy();
     Worker worker =
         new Worker(
-            workerId,
-            time,
-            plugins,
-            config,
-            new FileOffsetBackingStore(),
-            clientConfigOverridePolicy);
+            workerId, time, plugins, config, resettableOffsetStore, clientConfigOverridePolicy);
     this.herder = new StandaloneHerder(worker, kafkaClusterId, clientConfigOverridePolicy);
     connectionString = advertisedUrl.toString() + herder.kafkaClusterId();
 
@@ -166,6 +164,10 @@ void deleteConnector(final String name) {
     }
   }
 
+  void resetOffsets() {
+    resettableOffsetStore.reset();
+  }
+
   void stop() {
     LOGGER.debug("Connect Standalone stop called");
     connect.stop();
@@ -177,4 +179,11 @@ class ConnectorConfigurationException extends RuntimeException {
       super(cause);
     }
   }
+
+  private static class ResettableOffsetStore extends MemoryOffsetBackingStore {
+
+    void reset() {
+      data.clear();
+    }
+  }
 }

src/integrationTest/java/com/mongodb/kafka/connect/embedded/EmbeddedKafka.java

@@ -36,7 +36,6 @@
 
 import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.apache.kafka.common.security.JaasUtils;
-import org.apache.kafka.connect.runtime.distributed.DistributedConfig;
 import org.apache.kafka.connect.runtime.standalone.StandaloneConfig;
 import org.apache.kafka.streams.integration.utils.KafkaEmbedded;
 import org.apache.kafka.test.TestCondition;
@@ -253,6 +252,10 @@ public void deleteSourceConnector() {
     }
   }
 
+  public void resetOffsets() {
+    connect.resetOffsets();
+  }
+
   private Properties effectiveBrokerConfigFrom(
       final Properties brokerConfig, final ZooKeeperEmbedded zookeeper) {
     final Properties effectiveConfig = new Properties();
@@ -273,7 +276,6 @@ private Properties effectiveBrokerConfigFrom(
   private Properties connectWorkerConfig() {
     Properties workerProps = new Properties();
     workerProps.put(StandaloneConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
-    workerProps.put(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG, "connect-offsets");
     workerProps.put(
         StandaloneConfig.KEY_CONVERTER_CLASS_CONFIG,
         "org.apache.kafka.connect.storage.StringConverter");
@@ -295,6 +297,7 @@ public void beforeAll(final ExtensionContext context) throws Exception {
 
   @Override
   public void afterEach(final ExtensionContext context) {
+    resetOffsets();
     deleteSinkConnector();
     deleteSourceConnector();
   }

src/integrationTest/java/com/mongodb/kafka/connect/mongodb/MongoKafkaTestCase.java

@@ -132,6 +132,7 @@ public int getMaxWireVersion() {
   }
 
   public void cleanUp() {
+    KAFKA.resetOffsets();
     getMongoClient()
         .listDatabaseNames()
         .into(new ArrayList<>())

src/integrationTest/java/com/mongodb/kafka/connect/source/MongoSourceTaskIntegrationTest.java

@@ -431,15 +431,18 @@ void testSourceGeneratesHeartbeats() {
               put(MongoSourceConfig.COLLECTION_CONFIG, coll.getNamespace().getCollectionName());
               put(MongoSourceConfig.HEARTBEAT_TOPIC_NAME_CONFIG, "heartBeatTopic");
               put(MongoSourceConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "1000");
-              put(MongoSourceConfig.POLL_MAX_BATCH_SIZE_CONFIG, "50");
+              put(MongoSourceConfig.POLL_MAX_BATCH_SIZE_CONFIG, "10");
             }
           };
 
       task.start(cfg);
 
+      insertMany(rangeClosed(1, 10), coll);
+      getNextResults(task).forEach(s -> assertNotEquals("heartBeatTopic", s.topic()));
+
       getNextResults(task).forEach(s -> assertEquals("heartBeatTopic", s.topic()));
 
-      insertMany(rangeClosed(1, 50), coll);
+      insertMany(rangeClosed(11, 20), coll);
       getNextResults(task).forEach(s -> assertNotEquals("heartBeatTopic", s.topic()));
     }
   }