update in application.*.conf and clean up

Author: Heng Cui

candyfloss/src/main/resources/application.dev.conf

@@ -13,13 +13,15 @@ kafka {
   replication.factor = 2
   num.stream.threads = 2
   probing.rebalance.interval.ms = 120000
-  # we may need to tune "metadata.max.age.ms" and "partition.assignment.strategy" for pattern subscription.
-  # metadata.max.age.ms = 30000
+
+  # the following settings could be relevant for behaviors when having multiple topics subscribed.
   # partition.assignment.strategy = "org.apache.kafka.clients.consumer.CooperativeStickyAssignor"
+  # max.partition.fetch.bytes = 128000
+  # metadata.max.age.ms = 300000
 }
 kstream {
-  input.topic.pattern = "dev.*\\.device-.*-raw"
-  input.topic.name = dev.device-json-raw-input
+  input.topic.pattern = "dev\\.device-yang(-[^-]+)?-raw"
+  input.topic.name = "null"
   discard.topic.name = dev.candyfloss-processing-discard # All messages that didn't match any step in the pipeline
   dlq.topic.name = dev.candyfloss-processing--dlq # Messages that encountered a Java exception during the processing
   normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes

…main/resources/application.ietf.dev.conf …/main/resources/application.itm.dev.confcandyfloss/src/main/resources/application.ietf.dev.conf renamed to candyfloss/src/main/resources/application.itm.dev.conf candyfloss/src/main/resources/application.ietf.dev.conf renamed to candyfloss/src/main/resources/application.itm.dev.conf

@@ -1,12 +1,12 @@
 kafka {
   application.id = "daisy.dev.candyfloss"
   group.id = "daisy.dev.candyfloss-1"
-  bootstrap.servers = "kafka.sbd.corproot.net:9093"
+  bootstrap.servers = "localhost:9092"
   acks = 1
   linger.ms = 5
   processing.guarantee = at_least_once
   enable.idempotence = false
-  schema.registry.url = "https://schema-registry.zhh.sbd.corproot.net"
+  schema.registry.url = "https://localhost:8081"
   key.serializer = "org.apache.kafka.common.serialization.StringSerializer"
   value.serializer = "org.apache.kafka.common.serialization.StringSerializer"
   metrics.recording.level = DEBUG
@@ -15,6 +15,10 @@ kafka {
   num.stream.threads = 2
   auto.offset.reset = latest
   probing.rebalance.interval.ms = 120000
+
+  # partition.assignment.strategy = "org.apache.kafka.clients.consumer.CooperativeStickyAssignor"
+  # max.partition.fetch.bytes = 128000
+  # metadata.max.age.ms = 300000
 }
 kstream {
   input.topic.name = daisy.dev.device-json-raw
@@ -27,7 +31,7 @@ kstream {
   state.store.long.counter.wrap.limit = 10000000
   state.store.long.counter.time.ms = 300000 # allow max 5 min for counter wrap around otherwise it's a reset
   state.store.delete.scan.frequency.days = 7 # how often to trigger scanning for old unused counters and delete them
-  pre.transform = pre-transform.ietf.dev.json
+  pre.transform = pre-transform.itm.dev.json
   pipeline = {
     # todo: customize ietf-telemetry-message pipeline here
     fano-interfaces {

candyfloss/src/main/resources/application.prod.conf

@@ -13,6 +13,10 @@ kafka {
   replication.factor = 2
   num.stream.threads = 2
   probing.rebalance.interval.ms = 120000
+
+  # partition.assignment.strategy = "org.apache.kafka.clients.consumer.CooperativeStickyAssignor"
+  # max.partition.fetch.bytes = 128000
+  # metadata.max.age.ms = 300000
 }
 kstream {
   input.topic.name = prod.device-json-raw-input

…in/resources/pre-transform.ietf.dev.json …ain/resources/pre-transform.itm.dev.jsoncandyfloss/src/main/resources/pre-transform.ietf.dev.json renamed to candyfloss/src/main/resources/pre-transform.itm.dev.json candyfloss/src/main/resources/pre-transform.ietf.dev.json renamed to candyfloss/src/main/resources/pre-transform.itm.dev.json

@@ -25,9 +25,9 @@ protected void test(String applicationConfigFileName, String env, String name, P
   protected static Stream<Arguments> providerForAllYangModels()
       throws IOException, URISyntaxException {
     var outLegacy = providerForYangModels("legacy");
-    var outIetf = providerForYangModels("ietf");
+    var outItm = providerForYangModels("itm");
 
-    return Stream.concat(outLegacy, outIetf);
+    return Stream.concat(outLegacy, outItm);
   }
 
   /*** Discover YANG model test cases and provide them as arguments to the parametrized test */
@@ -38,7 +38,7 @@ protected static Stream<Arguments> providerForYangModels(String profile)
     // Create a map from the discovered folders in the `test/resources/deployment` folder
     // The key is the sub-folder name and the value is a full path to the sub folder
     String deploymentFolder =
-        profile.equals("ietf") ? "deployment-ietf-telemetry-message" : "deployment";
+        profile.equals("itm") ? "deployment-ietf-telemetry-message" : "deployment";
     var yangModels =
         Files.list(Path.of(Objects.requireNonNull(loader.getResource(deploymentFolder)).toURI()))
             .filter(Files::isDirectory)
@@ -57,7 +57,7 @@ protected static Stream<Arguments> providerForYangModels(String profile)
       var testFixtures = Files.list(path).filter(Files::isDirectory).toList();
       for (var testPath : testFixtures) {
         for (var env : List.of("dev", "test", "prod")) {
-          String fileNamePrefix = profile.equals("ietf") ? "application.ietf." : "application.";
+          String fileNamePrefix = profile.equals("itm") ? "application.itm." : "application.";
 
           String applicationConfig = fileNamePrefix + env + ".conf";
           if (loader.getResource(applicationConfig) != null)