@W-19993179 enable dynamic app name for kinesis consumer

Author: sholavanalli

carbonj.service/cj-deploy.sh

@@ -0,0 +1,31 @@
+#!/bin/bash
+
+# Shards 18-31 are part of the v2-B sidecar migration.
+# We are deploying to them with a 11 min delay.
+for i in {18..31}
+do
+  echo "Upgrading carbonj-p${i}..."
+  helm upgrade -f ./values-prd-v2-shard${i}.yaml carbonj-p${i} ./ -n carbonj
+  echo "Upgrade of carbonj-p${i} complete."
+  if [ $i -lt 31 ]; then
+    echo "Waiting for 11 minutes before next upgrade..."
+    
+    POD1="carbonj-p${i}-0"
+    POD2="carbonj-p${i}-1"
+
+    # Wait for approximately 11 minutes (22 * 30 seconds = 660 seconds), checking status periodically.
+    for j in {1..22}; do
+        echo "--- Status check ${j}/22 for shard p${i} ---"
+        kubectl get pod ${POD1} -n carbonj
+        kubectl get pod ${POD2} -n carbonj
+        echo "------------------------------------------------"
+        # If not the last check, sleep.
+        if [ $j -lt 22 ]; then
+            sleep 30
+        fi
+    done
+    echo "Finished 11 minute wait for shard p${i}."
+  fi
+done
+
+echo "All upgrades complete."

carbonj.service/config/kinesis-test-stream-consumer.conf

@@ -0,0 +1 @@
+foo=bar

carbonj.service/src/main/java/com/demandware/carbonj/service/engine/Consumers.java

@@ -19,13 +19,20 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.InetAddress;
-import java.net.UnknownHostException;
 import java.util.Date;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.net.UnknownHostException;
+import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
+import software.amazon.awssdk.services.dynamodb.model.ListTablesRequest;
+import software.amazon.awssdk.services.dynamodb.model.ListTablesResponse;
+import software.amazon.awssdk.services.dynamodb.model.DeleteTableRequest;
 
 public class Consumers {
 
@@ -149,6 +156,16 @@ private void reconfigureConsumers(Set<String> newRules, Set<String> currentRules
                     log.error(e.getMessage(), e);
                 }
 
+                // Optionally add dynamic suffix to the KCL application name
+                if (kinesisConfig.isAppNameDynamicSuffixEnabled()) {
+                    String suffix = resolveDynamicSuffix(kinesisConfig.getAppNameDynamicSuffixFormat());
+                    kinesisApplicationName = kinesisApplicationName + suffix;
+                    // Optionally kick off async cleanup of old lease tables
+                    if (kinesisConfig.isCleanupOldLeaseTablesEnabled()) {
+                        asyncCleanupOldLeaseTables(getKinesisApplicationName(kinesisStreamName, hostName, carbonjEnv));
+                    }
+                }
+
                 Counter initRetryCounter = metricRegistry.counter(MetricRegistry.name("kinesis.consumer." + kinesisStreamName + ".initRetryCounter"));
                 KinesisConsumer kinesisConsumer = new KinesisConsumer(metricRegistry, pointProcessor, recoveryPointProcessor, kinesisStreamName,
                         kinesisApplicationName, kinesisConfig, checkPointMgr, initRetryCounter, kinesisConsumerRegion, kinesisConsumerTracebackMinutes);
@@ -185,6 +202,50 @@ private String getKinesisApplicationName(String streamName, String hostName, Str
         return streamName + "-" + hostName + "-" + carbonjEnv;
     }
 
+    private String resolveDynamicSuffix(String format) {
+        String suffix = format;
+        try {
+            suffix = suffix.replace("{epoch}", Long.toString(System.currentTimeMillis()))
+                           .replace("{hostname}", InetAddress.getLocalHost().getHostName())
+                           .replace("{uuid}", java.util.UUID.randomUUID().toString());
+        } catch (Throwable t) {
+            // fallback minimal suffix
+            suffix = "-" + System.currentTimeMillis();
+        }
+        if (!suffix.startsWith("-")) {
+            suffix = "-" + suffix;
+        }
+        return suffix;
+    }
+
+    private void asyncCleanupOldLeaseTables(String baseApplicationName) {
+        ExecutorService es = Executors.newSingleThreadExecutor(r -> {
+            Thread t = new Thread(r, "kcl-lease-cleanup");
+            t.setDaemon(true);
+            return t;
+        });
+        es.submit(() -> {
+            try (DynamoDbClient ddb = DynamoDbClient.builder().build()) {
+                ListTablesResponse resp = ddb.listTables(ListTablesRequest.builder().build());
+                for (String table : resp.tableNames()) {
+                    // KCL v2/v3 table name prefix is usually application name; be conservative
+                    if (table.startsWith(baseApplicationName) && !table.equals(baseApplicationName)) {
+                        try {
+                            ddb.deleteTable(DeleteTableRequest.builder().tableName(table).build());
+                            log.info("Submitted async deletion for old KCL lease/checkpoint table {}", table);
+                        } catch (Throwable t) {
+                            log.warn("Failed to delete table {}: {}", table, t.getMessage());
+                        }
+                    }
+                }
+            } catch (Throwable t) {
+                log.warn("KCL lease cleanup encountered an error: {}", t.getMessage());
+            }
+        });
+        es.shutdown();
+        try { es.awaitTermination(1, TimeUnit.SECONDS); } catch (InterruptedException ignored) {}
+    }
+
     private void close(Set<String> consumerSet) {
         if (null == consumerSet) {
             return;

carbonj.service/src/main/java/com/demandware/carbonj/service/engine/KinesisConfig.java

@@ -29,11 +29,18 @@ public class KinesisConfig {
     private final int maxRecords;
     private final boolean aggregationEnabled;
 
+    // Controls dynamic application name behavior and cleanup of old lease tables
+    private final boolean appNameDynamicSuffixEnabled;
+    private final String appNameDynamicSuffixFormat;
+    private final boolean cleanupOldLeaseTablesEnabled;
+
     public KinesisConfig(boolean kinesisConsumerEnabled, boolean recoveryEnabled,
                          long recoveryIdleTimeMillis, long checkPointIntervalMillis, long retryTimeInMillis,
                          int recoveryThreads, Path checkPointDir, int initRetryTimeInSecs,
                          int leaseExpirationTimeInSecs, String recoveryProvider, int gapsTableProvThroughput,
-                         int maxRecords, boolean aggregationEnabled) {
+                         int maxRecords, boolean aggregationEnabled,
+                         boolean appNameDynamicSuffixEnabled, String appNameDynamicSuffixFormat,
+                         boolean cleanupOldLeaseTablesEnabled) {
         this.kinesisConsumerEnabled = kinesisConsumerEnabled;
         this.recoveryEnabled = recoveryEnabled;
         this.recoveryIdleTimeMillis = recoveryIdleTimeMillis;
@@ -46,6 +53,9 @@ public KinesisConfig(boolean kinesisConsumerEnabled, boolean recoveryEnabled,
         this.gapsTableProvisionedThroughput = gapsTableProvThroughput;
         this.maxRecords = maxRecords;
         this.aggregationEnabled = aggregationEnabled;
+        this.appNameDynamicSuffixEnabled = appNameDynamicSuffixEnabled;
+        this.appNameDynamicSuffixFormat = appNameDynamicSuffixFormat;
+        this.cleanupOldLeaseTablesEnabled = cleanupOldLeaseTablesEnabled;
 
         if( recoveryProvider.equalsIgnoreCase("dynamodb")) {
             this.recoveryProvider = KinesisRecoveryProvider.DYNAMODB;
@@ -102,4 +112,16 @@ public int getMaxRecords() {
     public boolean isAggregationEnabled() {
         return aggregationEnabled;
     }
+
+    public boolean isAppNameDynamicSuffixEnabled() {
+        return appNameDynamicSuffixEnabled;
+    }
+
+    public String getAppNameDynamicSuffixFormat() {
+        return appNameDynamicSuffixFormat;
+    }
+
+    public boolean isCleanupOldLeaseTablesEnabled() {
+        return cleanupOldLeaseTablesEnabled;
+    }
 }

carbonj.service/src/main/java/com/demandware/carbonj/service/engine/cfgKinesis.java

@@ -55,12 +55,24 @@ public class cfgKinesis
     @Value( "${aggregation.enabled:true}" )
     private boolean aggregationEnabled;
 
+    // Optional: dynamically suffix the KCL application name and cleanup old lease tables
+    @Value( "${kinesis.consumer.appName.dynamicSuffix.enabled:false}" )
+    private boolean appNameDynamicSuffixEnabled;
+
+    // Supports tokens: {epoch}, {uuid}, {hostname}
+    @Value( "${kinesis.consumer.appName.dynamicSuffix.format:-{epoch}}" )
+    private String appNameDynamicSuffixFormat;
+
+    @Value( "${kinesis.consumer.cleanupOldLeaseTables.enabled:false}" )
+    private boolean cleanupOldLeaseTablesEnabled;
+
     @Bean
     KinesisConfig kinesisConfig()
     {
         return new KinesisConfig(kinesisConsumerEnabled, recoveryEnabled, recoveryIdleTimeInMillis,
                 checkPointIntervalMillis, retryTimeInMillis, recoveryThreads, Paths.get(checkPointDir),
                 initRetryTimeInSecs, leaseExpirationTimeInSecs, recoveryProvider, gapsTableProvisionedThroughput,
-                maxRecords, aggregationEnabled);
+                maxRecords, aggregationEnabled,
+                appNameDynamicSuffixEnabled, appNameDynamicSuffixFormat, cleanupOldLeaseTablesEnabled);
     }
 }