adding the forceSync flag for handling conflicting topic configs

CHANGELOG.md

@@ -7,6 +7,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ## [0.5.1] - SNAPSHOT
 ### Added
 - Updated `README` with Expedia Group stream registry announcement (#155) 
+- Added the `forceSync` flag for handling conflicting topic configs while upserting stream (#114)
 
 ### Changed
 - Updated mkdocs.yml to `expediagroup` (#168)

core/src/main/java/com/homeaway/streamplatform/streamregistry/model/Stream.java

@@ -133,6 +133,13 @@ public class Stream {
     @ApiModelProperty(example = "3")
     int replicationFactor;
 
+    /**
+     * Should the provided topic config be forcibly synced with the existing, underlying topic (if not the same)?
+     */
+    @ApiModelProperty(example = "false")
+    @Builder.Default
+    Boolean forceSync = false;
+
     @JsonPOJOBuilder(withPrefix = "")
     public static final class StreamBuilder {
     }

core/src/main/java/com/homeaway/streamplatform/streamregistry/provider/InfraManager.java

@@ -98,11 +98,12 @@ public interface InfraManager {
      * @param replicationFactor replicationFactor for each of those topics
      * @param topicConfig topic config to use for each of these topics
      * @param isNewStream whether or not this invocation results from existing or new stream in stream registry.
+     * @param forceSync whether or not the provided topic config be forcibly synced with the existing, underlying topic (if not the same).
      * @throws StreamCreationException  when Stream could not be created in the underlying infrastructure for following reasons
      *      a) Input Configs and the existing configs does not match for a new Stream on-boarded to StreamRegistry,
      *      but already available in the infrastructure.
      */
-    void upsertTopics(Collection<String> topics, int partitions, int replicationFactor, Properties topicConfig, boolean isNewStream)
+    void upsertTopics(Collection<String> topics, int partitions, int replicationFactor, Properties topicConfig, boolean isNewStream, boolean forceSync)
         throws StreamCreationException;
 
 }

core/src/main/java/com/homeaway/streamplatform/streamregistry/service/impl/StreamServiceImpl.java

@@ -182,7 +182,7 @@ private void upsertTopics(Stream stream, String vpc, boolean isNewStream) throws
         topicConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
         topicConfig.put(KafkaProducerConfig.ZOOKEEPER_QUORUM, zkConnect);
 
-        infraManager.upsertTopics(Collections.singleton(stream.getName()), stream.getPartitions(), stream.getReplicationFactor(), topicConfig, isNewStream);
+        infraManager.upsertTopics(Collections.singleton(stream.getName()), stream.getPartitions(), stream.getReplicationFactor(), topicConfig, isNewStream, stream.getForceSync());
         log.info("Topic {} created/updated at {}", stream.getName(), bootstrapServer);
     }
 

core/src/test/java/com/homeaway/streamplatform/streamregistry/resource/InfraManagerImplStub.java

@@ -49,7 +49,7 @@ public Optional<ClusterValue> getClusterByKey(ClusterKey key) {
     }
 
     @Override
-    public void upsertTopics(Collection<String> topics, int partitions, int replicationFactor, Properties topicConfig, boolean isNewStream) {
+    public void upsertTopics(Collection<String> topics, int partitions, int replicationFactor, Properties topicConfig, boolean isNewStream, boolean forceSync) {
     }
 
     @Override

infra-provider-kafka/src/main/java/com/homeaway/streamplatform/streamregistry/provider/impl/KafkaInfraManager.java

@@ -180,7 +180,7 @@ public void upsertCluster(ClusterKey clusterKey, ClusterValue clusterValue) {
     }
 
     @Override
-    public void upsertTopics(Collection<String> topics, int partitions, int replicationFactor, Properties properties, boolean isNewStream)
+    public void upsertTopics(Collection<String> topics, int partitions, int replicationFactor, Properties properties, boolean isNewStream, boolean forceSync)
         throws StreamCreationException {
         // TODO - Cannot guarantee against race conditions... should probably move to event-source paradigm to
         //      protect against this (and maybe employ optimistic locking for extra safety).
@@ -202,7 +202,7 @@ public void upsertTopics(Collection<String> topics, int partitions, int replicat
             List<String> topicsToCreate = partitionMaps.get(false);
 
             // update any topics that are necessary
-            updateTopics(zkUtils, topicsToUpdate, topicConfigMap, isNewStream);
+            updateTopics(zkUtils, topicsToUpdate, topicConfigMap, isNewStream, forceSync);
 
             // now create any topics that were necessary to create this run
             createTopics(zkUtils, topicsToCreate, partitions, replicationFactor, topicConfigMap);
@@ -233,7 +233,7 @@ private boolean topicExists(ZkUtils zkUtils, String topic) {
         return topicExists;
     }
 
-    void updateTopics(ZkUtils zkUtils, List<String> topicsToUpdate, Map<String, String> topicConfigMap, boolean isStreamNotAvailableInStreamRegistryDB)
+    void updateTopics(ZkUtils zkUtils, List<String> topicsToUpdate, Map<String, String> topicConfigMap, boolean isStreamNotAvailableInStreamRegistryDB, boolean forceSync)
         throws StreamCreationException {
         for (String topic : topicsToUpdate) {
             // update topic
@@ -254,8 +254,12 @@ void updateTopics(ZkUtils zkUtils, List<String> topicsToUpdate, Map<String, Stri
             // to exactly match downstream config when the stream-registry has not "onboarded" existing topic
             // for the first time.
 
-            // TODO Alternatively we can add a forceSync=true flag, ignoring any user provided info, and only updating SR with the underlying settings
-            //      We should probably do forceSync=true anyway, as it provides a simple way to keep things in sync (#114)
+            // Added a forceSync=true flag, ignoring any user provided info, and only updating SR with the underlying settings (#114)
+            if (forceSync) {
+                // NOTHING TO DO!
+                log.info("topic config mismatch for {} ignored. Input config={}, actual (retained) config={}", topic, topicConfigMap, actualTopicConfig);
+                continue;
+            }
             if (isStreamNotAvailableInStreamRegistryDB) {
                 throw new StreamCreationException(String.format("Error: Input configs=%s and actual configs=%s are not same for topic=%s",
                     topicConfigMap, actualTopicConfig, topic));

infra-provider-kafka/src/test/java/com/homeaway/streamplatform/streamregistry/provider/impl/KafkaInfraManagerTest.java

@@ -97,7 +97,7 @@ public void testUpsertTopicsForNewStream() throws StreamCreationException {
         when(AdminUtils.topicExists(zkUtils, TOPIC)).thenReturn(true);
 
         //New Stream
-        kafkaManager.upsertTopics(Collections.singleton(TOPIC), PARTITIONS, REPLICATION_FACTOR, PROPS, true);
+        kafkaManager.upsertTopics(Collections.singleton(TOPIC), PARTITIONS, REPLICATION_FACTOR, PROPS, true, false);
 
         // expecting an exception to be thrown because topic exists but request doesn't match the config
     }
@@ -110,7 +110,7 @@ public void testUpsertTopicsForExistingStreamWithMatchingConfig() throws StreamC
         KafkaInfraManager kafkaManagerSpy = spy(kafkaManager);
 
         // Existing Stream, but PROPS match!! should not have an exception
-        kafkaManagerSpy.upsertTopics(Collections.singleton(TOPIC), PARTITIONS, REPLICATION_FACTOR, TOPIC_WITH_CNXN_PROPS, true);
+        kafkaManagerSpy.upsertTopics(Collections.singleton(TOPIC), PARTITIONS, REPLICATION_FACTOR, TOPIC_WITH_CNXN_PROPS, true, false);
 
         // verify change topic DOES NOT HAPPEN because props match
         verifyStatic(AdminUtils.class, times(0));
@@ -129,7 +129,7 @@ public void testUpsertTopicsForExistingStream() throws StreamCreationException {
         KafkaInfraManager kafkaManagerSpy = spy(kafkaManager);
 
         //Existing Stream
-        kafkaManagerSpy.upsertTopics(Collections.singleton(TOPIC), PARTITIONS, REPLICATION_FACTOR, PROPS, false);
+        kafkaManagerSpy.upsertTopics(Collections.singleton(TOPIC), PARTITIONS, REPLICATION_FACTOR, PROPS, false, false);
 
         //verify change topic happens because isNewStream=false
         verifyStatic(AdminUtils.class, times(1));
@@ -147,7 +147,7 @@ public void testUpsertTopicsForNewTopic() throws StreamCreationException {
         KafkaInfraManager kafkaManagerSpy = spy(kafkaManager);
 
         // Not existing Stream
-        kafkaManagerSpy.upsertTopics(Collections.singleton(TOPIC), PARTITIONS, REPLICATION_FACTOR, PROPS, true);
+        kafkaManagerSpy.upsertTopics(Collections.singleton(TOPIC), PARTITIONS, REPLICATION_FACTOR, PROPS, true, false);
 
         //verify create topic happens when requested topic does not exist
         verifyStatic(AdminUtils.class, times(0));
@@ -164,7 +164,7 @@ public void testUpsertTopicsForNewTopicExistsInSR() throws StreamCreationExcepti
         KafkaInfraManager kafkaManagerSpy = spy(kafkaManager);
 
         //Existing Stream
-        kafkaManagerSpy.upsertTopics(Collections.singleton(TOPIC), PARTITIONS, REPLICATION_FACTOR, PROPS, false);
+        kafkaManagerSpy.upsertTopics(Collections.singleton(TOPIC), PARTITIONS, REPLICATION_FACTOR, PROPS, false, false);
 
         // note: this is a weird case, because somehow the stream exists in SR, but the underlying topic does NOT
         // might want to consider this corner case a bit more. Currently the behavior honors the request and creates the topic
@@ -176,4 +176,46 @@ public void testUpsertTopicsForNewTopicExistsInSR() throws StreamCreationExcepti
         verifyStatic(AdminUtils.class, times(1));
         AdminUtils.createTopic(zkUtils, TOPIC, PARTITIONS, REPLICATION_FACTOR, FILTERED_PROPS, RackAwareMode.Enforced$.MODULE$);
     }
+
+    @Test(expected = StreamCreationException.class)
+    public void testUpsertTopicsForExistingStreamWithMismatchingConfigForceSyncDisabled() throws StreamCreationException {
+        // Mock it as an existing topic
+        when(AdminUtils.topicExists(zkUtils, TOPIC)).thenReturn(true);
+
+        KafkaInfraManager kafkaManagerSpy = spy(kafkaManager);
+
+        // Existing Stream, but PROPS match!! should not have an exception
+        Properties prop = new Properties();
+        prop.put("k1", "v1");
+        kafkaManagerSpy.upsertTopics(Collections.singleton(TOPIC), PARTITIONS, REPLICATION_FACTOR, prop, true, false);
+
+        // verify change topic DOES NOT HAPPEN because props match
+        verifyStatic(AdminUtils.class, times(0));
+        AdminUtils.changeTopicConfig(zkUtils, TOPIC, TOPIC_PROPS);
+
+        // verify create topic DOES NOT HAPPEN because props match
+        verifyStatic(AdminUtils.class, times(0));
+        AdminUtils.createTopic(zkUtils, TOPIC, PARTITIONS, REPLICATION_FACTOR, TOPIC_PROPS, RackAwareMode.Enforced$.MODULE$);
+    }
+
+    @Test
+    public void testUpsertTopicsForExistingStreamWithMismatchingConfigForceSyncEnabled() throws StreamCreationException {
+        // Mock it as an existing topic
+        when(AdminUtils.topicExists(zkUtils, TOPIC)).thenReturn(true);
+
+        KafkaInfraManager kafkaManagerSpy = spy(kafkaManager);
+
+        // Existing Stream, but PROPS match!! should not have an exception
+        Properties prop = new Properties();
+        prop.put("MAX_MESSAGE_BYTES_CONFIG", "64000");
+        kafkaManagerSpy.upsertTopics(Collections.singleton(TOPIC), PARTITIONS, REPLICATION_FACTOR, prop, true, true);
+
+        // verify change topic DOES NOT HAPPEN because props match
+        verifyStatic(AdminUtils.class, times(0));
+        AdminUtils.changeTopicConfig(zkUtils, TOPIC, TOPIC_PROPS);
+
+        // verify create topic DOES NOT HAPPEN because props match
+        verifyStatic(AdminUtils.class, times(0));
+        AdminUtils.createTopic(zkUtils, TOPIC, PARTITIONS, REPLICATION_FACTOR, TOPIC_PROPS, RackAwareMode.Enforced$.MODULE$);
+    }
 }

schema-library-events/src/main/avro/stream.avdl

@@ -196,6 +196,12 @@ protocol StreamRegistryProtocol {
     * Replication Factor
     */
     int replicationFactor = 3;
+
+    /**
+    * Should the provided topic config be forcibly synced with the existing, underlying topic (if not the same)?
+    */
+    boolean forceSync = false;
+
   }
 
   record AvroStreamKey {