W-17312191: Make carbonj.env as part of KCL application name

peterzxu-crm · peterzxu-crm · commit 447bbc807f8d · 2025-09-05T10:48:32.000-04:00
diff --git a/carbonj.service/src/main/java/com/demandware/carbonj/service/engine/Consumers.java b/carbonj.service/src/main/java/com/demandware/carbonj/service/engine/Consumers.java
@@ -50,9 +50,11 @@ public class Consumers {
 
     private final File indexNameSyncDir;
 
+    private final String carbonjEnv;
+
     Consumers(MetricRegistry metricRegistry, PointProcessor pointProcessor, PointProcessor recoveryPointProcessor, File rulesFile,
               KinesisConfig kinesisConfig, CheckPointMgr<Date> checkPointMgr, String kinesisConsumerRegion,
-              NamespaceCounter namespaceCounter, File indexNameSyncDir) {
+              NamespaceCounter namespaceCounter, File indexNameSyncDir, String carbonjEnv) {
 
         this.metricRegistry = metricRegistry;
         this.pointProcessor = pointProcessor;
@@ -62,8 +64,9 @@ public class Consumers {
         this.kinesisConsumerRegion = kinesisConsumerRegion;
         this.namespaceCounter = namespaceCounter;
         this.indexNameSyncDir = indexNameSyncDir;
-        consumers = new ConcurrentHashMap<>();
-        consumerRules = new ConsumerRules(rulesFile);
+        this.consumers = new ConcurrentHashMap<>();
+        this.consumerRules = new ConsumerRules(rulesFile);
+        this.carbonjEnv = carbonjEnv;
         reload();
     }
 
@@ -119,7 +122,7 @@ private void reconfigureConsumers(Set<String> newRules, Set<String> currentRules
             if (consumerName.startsWith("kinesis:")) {
 
                 String kinesisStreamName = consumerName.substring(("kinesis:".length()));
-                String kinesisApplicationName = getKinesisApplicationName(kinesisStreamName, hostName);
+                String kinesisApplicationName = getKinesisApplicationName(kinesisStreamName, hostName, carbonjEnv);
                 String consumerCfgFile = "config/kinesis-" + kinesisStreamName + "-consumer.conf";
 
                 try {
@@ -140,7 +143,7 @@ private void reconfigureConsumers(Set<String> newRules, Set<String> currentRules
 
                 Counter initRetryCounter = metricRegistry.counter(MetricRegistry.name("kinesis.consumer." + kinesisStreamName + ".initRetryCounter"));
                 KinesisConsumer kinesisConsumer = new KinesisConsumer(metricRegistry, pointProcessor, recoveryPointProcessor, kinesisStreamName,
-                        kinesisApplicationName, kinesisConfig, checkPointMgr, initRetryCounter, kinesisConsumerRegion);
+                        kinesisApplicationName, kinesisConfig, checkPointMgr, initRetryCounter, kinesisConsumerRegion, carbonjEnv);
                 log.info(String.format("New Consumer created with name %s", kinesisStreamName));
                 newConsumers.add(consumerName);
                 consumers.put(consumerName, kinesisConsumer);
@@ -170,8 +173,8 @@ private String getHostName() {
         }
     }
 
-    private String getKinesisApplicationName(String streamName, String hostName)  {
-        return streamName + "-" + hostName;
+    private String getKinesisApplicationName(String streamName, String hostName, String carbonjEnv)  {
+        return streamName + "-" + hostName + "-" + carbonjEnv;
     }
 
     private void close(Set<String> consumerSet) {
diff --git a/carbonj.service/src/main/java/com/demandware/carbonj/service/engine/cfgCarbonJ.java b/carbonj.service/src/main/java/com/demandware/carbonj/service/engine/cfgCarbonJ.java
@@ -207,6 +207,9 @@ public class cfgCarbonJ
     @Value("${metrics.store.sync.secondary.db:false}")
     private boolean syncSecondaryDb;
 
+    @Value("${carbonj.env:dev}")
+    private String carbonjEnv;
+
     @Bean
     public RestTemplate restTemplate() {
         return new RestTemplate();
@@ -424,7 +427,7 @@ Consumers consumer( PointProcessor pointProcessor,
             File rulesFile = locateConfigFile( serviceDir, consumerRulesFile );
             Consumers consumer = new Consumers( metricRegistry, pointProcessor, recoveryPointProcessor, rulesFile,
                     kinesisConfig, checkPointMgr, kinesisConsumerRegion,
-                    nsCounter, dataDir == null ? null : new File(dataDir, "index-name-sync"));
+                    nsCounter, dataDir == null ? null : new File(dataDir, "index-name-sync"), carbonjEnv);
             s.scheduleWithFixedDelay( consumer::reload, 15, 30, TimeUnit.SECONDS );
             if (syncSecondaryDb) {
                 s.scheduleWithFixedDelay( consumer::syncNamespaces, 60, 60, TimeUnit.SECONDS );
@@ -468,8 +471,7 @@ Void checkExpiredNamespaces( ScheduledExecutorService s, NamespaceCounter ns )
     {
         if ( runInactiveNamespaceCheckEverySeconds > 0 )
         {
-            log.info( String.format( "scheduling removal of expired namespace counters to run every %s sec",
-                            runInactiveNamespaceCheckEverySeconds ) );
+            log.info("scheduling removal of expired namespace counters to run every {} sec", runInactiveNamespaceCheckEverySeconds);
             s.scheduleWithFixedDelay( ns::removeInactive, 120, runInactiveNamespaceCheckEverySeconds,
                             TimeUnit.SECONDS );
         }
@@ -506,16 +508,14 @@ Void checkExpiredNamespaces( ScheduledExecutorService s, NamespaceCounter ns )
     NettyChannel lineProtocolChannel( NettyServer netty, InputQueue r )
     {
         lineProtocolTcpPort = ( lineProtocolTcpPort == -1 ) ? jettyPort + 2 : lineProtocolTcpPort;
-        return netty.bind( lineProtocolTcpHost, lineProtocolTcpPort, new ChannelInitializer<SocketChannel>()
-        {
-            @Override public void initChannel(@SuppressWarnings("NullableProblems") SocketChannel ch )
-            {
-                if ( log.isDebugEnabled() )
-                {
-                    log.debug( "accepted TCP line protocol from " + ch );
+        return netty.bind( lineProtocolTcpHost, lineProtocolTcpPort, new ChannelInitializer<>() {
+            @Override
+            public void initChannel(SocketChannel ch) {
+                if (log.isDebugEnabled()) {
+                    log.debug("accepted TCP line protocol from {}", ch);
                 }
-                ch.pipeline().addLast( new DelimiterBasedFrameDecoder( tcpBuff, Delimiters.lineDelimiter() ),
-                                new LineProtocolHandler( metricRegistry, r ) );
+                ch.pipeline().addLast(new DelimiterBasedFrameDecoder(tcpBuff, Delimiters.lineDelimiter()),
+                        new LineProtocolHandler(metricRegistry, r));
             }
         } );
     }
@@ -536,7 +536,7 @@ NettyChannel udpLineProtocolChannel( NettyServer netty, InputQueue r )
                                 NettyServer.udpMsgsReceived.mark();
                                 if ( log.isDebugEnabled() )
                                 {
-                                    log.debug( "accepted UDP line protocol from " + ctx.channel() );
+                                    log.debug("accepted UDP line protocol from {}", ctx.channel());
                                 }
                                 lp.process( msg.content() );
                             }
@@ -551,13 +551,13 @@ NettyChannel udpLineProtocolChannel( NettyServer netty, InputQueue r )
     @ConditionalOnProperty(name = "carbonj.relay", havingValue = "true", matchIfMissing = true)
     NettyChannel pickleProtocolChannel( NettyServer netty, InputQueue r )
     {
-        return netty.bind( "0.0.0.0", jettyPort + 3, new ChannelInitializer<SocketChannel>()
+        return netty.bind( "0.0.0.0", jettyPort + 3, new ChannelInitializer<>()
         {
-            @Override public void initChannel(@SuppressWarnings("NullableProblems") SocketChannel ch )
+            @Override public void initChannel(SocketChannel ch )
             {
                 if ( log.isDebugEnabled() )
                 {
-                    log.debug( "accepted pickle protocol from " + ch );
+                    log.debug("accepted pickle protocol from {}", ch);
                 }
 
                 ch.pipeline().addLast( new LengthFieldBasedFrameDecoder( pickleBuff, 0, 4 ) )
@@ -625,7 +625,7 @@ class Store:
     {
         for ( String profileName : environment.getActiveProfiles() )
         {
-            log.warn( "Currently active profile: " + profileName );
+            log.warn("Currently active profile: {}", profileName);
         }
     }
 
diff --git a/carbonj.service/src/test/java/com/demandware/carbonj/service/engine/TestConsumers.java b/carbonj.service/src/test/java/com/demandware/carbonj/service/engine/TestConsumers.java
@@ -38,7 +38,7 @@ public void testConsumers() throws Exception {
                 1, checkPointDir, 60, 60, "recoveryProvider", 1, 1, true);
         FileCheckPointMgr checkPointMgr = new FileCheckPointMgr(checkPointDir, 5);
         Consumers consumers = new Consumers(metricRegistry, new PointProcessorMock(), new PointProcessorMock(),
-                rulesFile, kinesisConfig, checkPointMgr, Region.US_EAST_1.id(), new NamespaceCounter(metricRegistry, 60), new File("/tmp/sync"));
+                rulesFile, kinesisConfig, checkPointMgr, Region.US_EAST_1.id(), new NamespaceCounter(metricRegistry, 60), new File("/tmp/sync"), "test");
         new KinesisRecordProcessorFactory(metricRegistry, new PointProcessorMock(), kinesisConfig, "test-stream", checkPointMgr);
         consumers.dumpStats();
         consumers.syncNamespaces();
@@ -54,7 +54,7 @@ public void testConsumersNoAggregation() throws Exception {
                 1, checkPointDir, 60, 60, "recoveryProvider", 1, 1, false);
         FileCheckPointMgr checkPointMgr = new FileCheckPointMgr(checkPointDir, 5);
         Consumers consumers = new Consumers(metricRegistry, new PointProcessorMock(), new PointProcessorMock(),
-                rulesFile, kinesisConfig, checkPointMgr, Region.US_EAST_1.id(), new NamespaceCounter(metricRegistry, 60), new File("/tmp/sync"));
+                rulesFile, kinesisConfig, checkPointMgr, Region.US_EAST_1.id(), new NamespaceCounter(metricRegistry, 60), new File("/tmp/sync"), "test");
         new KinesisRecordProcessorFactory(metricRegistry, new PointProcessorMock(), kinesisConfig, "test-stream", checkPointMgr);
         consumers.dumpStats();
         consumers.syncNamespaces();

Original file line number	Diff line number	Diff line change
`@@ -207,6 +207,9 @@ public class cfgCarbonJ`
`207`	`207`	`@Value("${metrics.store.sync.secondary.db:false}")`
`208`	`208`	`private boolean syncSecondaryDb;`
`209`	`209`
	`210`	`+ @Value("${carbonj.env:dev}")`
	`211`	`+ private String carbonjEnv;`
	`212`	`+`
`210`	`213`	`@Bean`
`211`	`214`	`public RestTemplate restTemplate() {`
`212`	`215`	`return new RestTemplate();`
`@@ -424,7 +427,7 @@ Consumers consumer( PointProcessor pointProcessor,`
`424`	`427`	`File rulesFile = locateConfigFile( serviceDir, consumerRulesFile );`
`425`	`428`	`Consumers consumer = new Consumers( metricRegistry, pointProcessor, recoveryPointProcessor, rulesFile,`
`426`	`429`	`kinesisConfig, checkPointMgr, kinesisConsumerRegion,`
`427`		`- nsCounter, dataDir == null ? null : new File(dataDir, "index-name-sync"));`
	`430`	`+ nsCounter, dataDir == null ? null : new File(dataDir, "index-name-sync"), carbonjEnv);`
`428`	`431`	`s.scheduleWithFixedDelay( consumer::reload, 15, 30, TimeUnit.SECONDS );`
`429`	`432`	`if (syncSecondaryDb) {`
`430`	`433`	`s.scheduleWithFixedDelay( consumer::syncNamespaces, 60, 60, TimeUnit.SECONDS );`
`@@ -468,8 +471,7 @@ Void checkExpiredNamespaces( ScheduledExecutorService s, NamespaceCounter ns )`
`468`	`471`	`{`
`469`	`472`	`if ( runInactiveNamespaceCheckEverySeconds > 0 )`
`470`	`473`	`{`
`471`		`- log.info( String.format( "scheduling removal of expired namespace counters to run every %s sec",`
`472`		`- runInactiveNamespaceCheckEverySeconds ) );`
	`474`	`+ log.info("scheduling removal of expired namespace counters to run every {} sec", runInactiveNamespaceCheckEverySeconds);`
`473`	`475`	`s.scheduleWithFixedDelay( ns::removeInactive, 120, runInactiveNamespaceCheckEverySeconds,`
`474`	`476`	`TimeUnit.SECONDS );`
`475`	`477`	`}`
`@@ -506,16 +508,14 @@ Void checkExpiredNamespaces( ScheduledExecutorService s, NamespaceCounter ns )`
`506`	`508`	`NettyChannel lineProtocolChannel( NettyServer netty, InputQueue r )`
`507`	`509`	`{`
`508`	`510`	`lineProtocolTcpPort = ( lineProtocolTcpPort == -1 ) ? jettyPort + 2 : lineProtocolTcpPort;`
`509`		`- return netty.bind( lineProtocolTcpHost, lineProtocolTcpPort, new ChannelInitializer<SocketChannel>()`
`510`		`- {`
`511`		`- @Override public void initChannel(@SuppressWarnings("NullableProblems") SocketChannel ch )`
`512`		`- {`
`513`		`- if ( log.isDebugEnabled() )`
`514`		`- {`
`515`		`- log.debug( "accepted TCP line protocol from " + ch );`
	`511`	`+ return netty.bind( lineProtocolTcpHost, lineProtocolTcpPort, new ChannelInitializer<>() {`
	`512`	`+ @Override`
	`513`	`+ public void initChannel(SocketChannel ch) {`
	`514`	`+ if (log.isDebugEnabled()) {`
	`515`	`+ log.debug("accepted TCP line protocol from {}", ch);`
`516`	`516`	`}`
`517`		`- ch.pipeline().addLast( new DelimiterBasedFrameDecoder( tcpBuff, Delimiters.lineDelimiter() ),`
`518`		`- new LineProtocolHandler( metricRegistry, r ) );`
	`517`	`+ ch.pipeline().addLast(new DelimiterBasedFrameDecoder(tcpBuff, Delimiters.lineDelimiter()),`
	`518`	`+ new LineProtocolHandler(metricRegistry, r));`
`519`	`519`	`}`
`520`	`520`	`} );`
`521`	`521`	`}`
`@@ -536,7 +536,7 @@ NettyChannel udpLineProtocolChannel( NettyServer netty, InputQueue r )`
`536`	`536`	`NettyServer.udpMsgsReceived.mark();`
`537`	`537`	`if ( log.isDebugEnabled() )`
`538`	`538`	`{`
`539`		`- log.debug( "accepted UDP line protocol from " + ctx.channel() );`
	`539`	`+ log.debug("accepted UDP line protocol from {}", ctx.channel());`
`540`	`540`	`}`
`541`	`541`	`lp.process( msg.content() );`
`542`	`542`	`}`
`@@ -551,13 +551,13 @@ NettyChannel udpLineProtocolChannel( NettyServer netty, InputQueue r )`
`551`	`551`	`@ConditionalOnProperty(name = "carbonj.relay", havingValue = "true", matchIfMissing = true)`
`552`	`552`	`NettyChannel pickleProtocolChannel( NettyServer netty, InputQueue r )`
`553`	`553`	`{`
`554`		`- return netty.bind( "0.0.0.0", jettyPort + 3, new ChannelInitializer<SocketChannel>()`
	`554`	`+ return netty.bind( "0.0.0.0", jettyPort + 3, new ChannelInitializer<>()`
`555`	`555`	`{`
`556`		`- @Override public void initChannel(@SuppressWarnings("NullableProblems") SocketChannel ch )`
	`556`	`+ @Override public void initChannel(SocketChannel ch )`
`557`	`557`	`{`
`558`	`558`	`if ( log.isDebugEnabled() )`
`559`	`559`	`{`
`560`		`- log.debug( "accepted pickle protocol from " + ch );`
	`560`	`+ log.debug("accepted pickle protocol from {}", ch);`
`561`	`561`	`}`
`562`	`562`
`563`	`563`	`ch.pipeline().addLast( new LengthFieldBasedFrameDecoder( pickleBuff, 0, 4 ) )`
`@@ -625,7 +625,7 @@ class Store:`
`625`	`625`	`{`
`626`	`626`	`for ( String profileName : environment.getActiveProfiles() )`
`627`	`627`	`{`
`628`		`- log.warn( "Currently active profile: " + profileName );`
	`628`	`+ log.warn("Currently active profile: {}", profileName);`
`629`	`629`	`}`
`630`	`630`	`}`
`631`	`631`