Ensure the heartbeat message schema is optional

A non optional schema breaks value conversion.

KAFKA-207 KAFKA-208

Author: rozza

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

@@ -44,6 +44,7 @@
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 
+import org.bson.BsonDocument;
 import org.bson.Document;
 
 import com.mongodb.client.MongoCollection;
@@ -308,6 +309,33 @@ void testSourceUsesHeartbeatsForOffsets() {
     }
   }
 
+  @Test
+  @DisplayName("Ensure Source heartbeats have a valid schema")
+  void testSourceHeartbeatsHaveValidSchema() {
+    assumeTrue(isGreaterThanFourDotZero());
+
+    MongoCollection<Document> coll = getAndCreateCollection();
+
+    String heartbeatTopic = "__HEARTBEAT_SCHEMA";
+
+    Properties sourceProperties = new Properties();
+    sourceProperties.put(MongoSourceConfig.DATABASE_CONFIG, coll.getNamespace().getDatabaseName());
+    sourceProperties.put(
+        MongoSourceConfig.COLLECTION_CONFIG, coll.getNamespace().getCollectionName());
+    sourceProperties.put(MongoSourceConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "1000");
+    sourceProperties.put(MongoSourceConfig.HEARTBEAT_TOPIC_NAME_CONFIG, heartbeatTopic);
+    sourceProperties.put("key.converter", "org.apache.kafka.connect.json.JsonConverter");
+    sourceProperties.put("key.converter.schemas.enable", "false");
+    sourceProperties.put("value.converter", "org.apache.kafka.connect.json.JsonConverter");
+    sourceProperties.put("value.converter.schemas.enable", "false");
+
+    addSourceConnector(sourceProperties);
+    BsonDocument heartbeat = getHeartbeat(heartbeatTopic);
+
+    assertTrue(heartbeat.get("key").isDocument());
+    assertTrue(heartbeat.get("value").isNull());
+  }
+
   @Test
   @DisplayName("Ensure Source provides friendly error messages for invalid pipelines")
   void testSourceHasFriendlyErrorMessagesForInvalidPipelines() {

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

@@ -40,6 +40,7 @@
 import org.apache.kafka.common.serialization.BytesDeserializer;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.connect.json.JsonDeserializer;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -329,6 +330,19 @@ public void assertProducedDocs(final List<Document> docs, final MongoCollection<
 
   private static final Deserializer<Bytes> BYTES_DESERIALIZER = new BytesDeserializer();
 
+  public BsonDocument getHeartbeat(final String topicName) {
+    return getProduced(
+            topicName,
+            new JsonDeserializer(),
+            new JsonDeserializer(),
+            c ->
+                BsonDocument.parse(
+                    format("{key: %s, value: %s}", c.key().textValue(), c.value().textValue())),
+            1,
+            1)
+        .get(0);
+  }
+
   public List<String> getProducedStrings(final String topicName, final int expectedSize) {
     return getProduced(
         topicName,
@@ -353,16 +367,6 @@ public <T> List<T> getProduced(
         maxRetryCount);
   }
 
-  public List<String> getProducedKeys(final MongoCollection<?> collection, final int expectedSize) {
-    return getProduced(
-        collection.getNamespace().getFullName(),
-        new MappingDeserializer<>(Bytes::toString),
-        BYTES_DESERIALIZER,
-        ConsumerRecord::key,
-        expectedSize,
-        DEFAULT_MAX_RETRIES);
-  }
-
   public static class MappingDeserializer<T> implements Deserializer<T> {
     private final Function<Bytes, T> mapper;
 

src/main/java/com/mongodb/kafka/connect/source/heartbeat/HeartbeatManager.java

@@ -85,7 +85,7 @@ public Optional<SourceRecord> heartbeat() {
                       heartbeatTopicName,
                       Schema.STRING_SCHEMA,
                       resumeToken,
-                      Schema.BYTES_SCHEMA,
+                      Schema.OPTIONAL_BYTES_SCHEMA,
                       null);
                 }
                 return null;