W-11492245: Fix CheckPointMgr null check for cjclient as Kinesis cons…

…umer
salesforce · Jul 24, 2023 · 3fea476 · 3fea476
1 parent 0a82d98
commit 3fea476
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 6 deletions.
diff --git a/carbonj.service/src/main/java/com/demandware/carbonj/service/engine/KinesisConsumer.java b/carbonj.service/src/main/java/com/demandware/carbonj/service/engine/KinesisConsumer.java
@@ -61,7 +61,7 @@ public KinesisConsumer(MetricRegistry metricRegistry, PointProcessor pointProces
         this.kinesisStreamName = kinesisStreamName;
         this.kinesisApplicationName = kinesisApplicationName;
         this.kinesisConfig = kinesisConfig;
-        this.checkPointMgr =  Preconditions.checkNotNull(checkPointMgr);
+        this.checkPointMgr = checkPointMgr;
         this.noOfRestarts = noOfRestarts;
         this.kinesisConsumerRegion = kinesisConsumerRegion;
         log.info("Kinesis consumer started");

diff --git a/...service/src/main/java/com/demandware/carbonj/service/engine/kinesis/cfgCheckPointMgr.java b/...service/src/main/java/com/demandware/carbonj/service/engine/kinesis/cfgCheckPointMgr.java
@@ -10,7 +10,12 @@
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder;
 import com.demandware.carbonj.service.accumulator.Accumulator;
 import com.demandware.carbonj.service.accumulator.cfgAccumulator;
-import com.demandware.carbonj.service.engine.*;
+import com.demandware.carbonj.service.engine.CheckPointMgr;
+import com.demandware.carbonj.service.engine.DynamoDbCheckPointMgr;
+import com.demandware.carbonj.service.engine.FileCheckPointMgr;
+import com.demandware.carbonj.service.engine.KinesisConfig;
+import com.demandware.carbonj.service.engine.cfgCarbonJ;
+import com.demandware.carbonj.service.engine.cfgKinesis;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -47,10 +52,13 @@ CheckPointMgr<Date> checkPointMgr(ScheduledExecutorService s, KinesisConfig kine
                                       @Autowired( required = false ) @Qualifier( "accumulator" ) Accumulator accu )
             throws Exception
     {
-        if ( !kinesisConfig.isKinesisConsumerEnabled() || accu == null )
+        if (!kinesisConfig.isKinesisConsumerEnabled()) {
+            log.warn("CheckPointMgr is disabled because kinesis consumer is disabled");
+            return null;
+        }
+        if (accu == null)
         {
-            log.debug( "CheckPointMgr is disabled because kinesis consumer is disabled or accumulator is null: Kinesis"
-                    + " consumer: {}, Accumulator: {}", kinesisConfig.isKinesisConsumerEnabled(), accu );
+            log.warn("CheckPointMgr is disabled because accumulator is null");
             return null;
         }
 
@@ -84,4 +92,4 @@ CheckPointMgr<Date> checkPointMgr(ScheduledExecutorService s, KinesisConfig kine
         }, 120, 60, TimeUnit.SECONDS );
         return checkPointMgr;
     }
-}
+}