PD-257019 Producer send and logging changes (#838)

* fix: fix send logic and exception handling changes

* added exception handling

* changed producer send logic to be synchronous

* removed redundant logs

* uncached_size for dedupq (#837)

* chore: rename queuetype as per convention

* chore: fix naming issue

---------

Co-authored-by: nabajyotiDash-hub <[email protected]>

Author: rishuyadavbv

queue/src/main/java/com/bazaarvoice/emodb/queue/core/AbstractQueueService.java

@@ -240,7 +240,7 @@ public void sendAll(Map<String, ? extends Collection<?>> messagesByQueue) {
         String queueType = determineQueueType();
         for (Map.Entry<String, Collection<String>> topicEntry : eventsByChannel.asMap().entrySet()) {
             String queueName= topicEntry.getKey();
-            String topic = "dsq-" + (("dedup".equals(queueType)) ?  "dedup-" + queueName : queueName);
+            String topic = "dsq-" + (("dedupq".equals(queueType)) ?  "dedup-" + queueName : queueName);
             // Check if the topic exists, if not create it and execute Step Function
             if (!adminService.createTopicIfNotExists(topic, TOPIC_PARTITION_COUNT, TOPIC_REPLICATION_FACTOR, queueType)) {
                 Map<String, String> parameters = fetchStepFunctionParameters();
@@ -456,8 +456,8 @@ private void startStepFunctionExecution(Map<String, String> parameters, String q
             // Create the timestamp
             String timestamp = String.valueOf(System.currentTimeMillis()); // Current time in milliseconds
 
-            // Check if queueType is "dedup" and prepend "D" to execution name if true
-            String executionName = (queueType.equalsIgnoreCase("dedup") ? "D_" : "") + queueName + "_" + timestamp;
+            // Check if queueType is "dedupq" and prepend "D" to execution name if true
+            String executionName = (queueType.equalsIgnoreCase("dedupq") ? "D_" : "") + queueName + "_" + timestamp;
 
             // Start the Step Function execution
             stepFunctionService.startExecution(stateMachineArn, inputPayload, executionName);
@@ -474,7 +474,7 @@ private void startStepFunctionExecution(Map<String, String> parameters, String q
      */
     private String determineQueueType() {
         if (_eventStore.getClass().getName().equals("com.bazaarvoice.emodb.event.dedup.DefaultDedupEventStore")) {
-            return "dedup";
+            return "dedupq";
         }
         return "queue";
     }

queue/src/main/java/com/bazaarvoice/emodb/queue/core/kafka/KafkaConfig.java

@@ -21,6 +21,7 @@
 import java.util.Map;
 import java.util.Properties;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 public class KafkaConfig {
     private static String bootstrapServersConfig;
@@ -33,41 +34,36 @@ public class KafkaConfig {
             .standard()
             .build();
 
-    private static final String DEFAULT_BOOTSTRAP_SERVERS =
-            "b-1.qaemodbpocmsk.q4panq.c10.kafka.us-east-1.amazonaws.com:9092," +
-                    "b-2.qaemodbpocmsk.q4panq.c10.kafka.us-east-1.amazonaws.com:9092";
-
 
     static {
         try {
             final String UNIVERSE = getUniverseFromEnv();
             // Load configurations from SSM during static initialization
+            String basePath = "/" + UNIVERSE + "/emodb/kafka/";
+            // Load configurations from SSM during static initialization
             Map<String, String> parameterValues = getParameterValues(
-                    Arrays.asList(
-                            "/" + UNIVERSE + "/emodb/kafka/batchSize",
-                            "/" + UNIVERSE + "/emodb/kafka/retries",
-                            "/" + UNIVERSE + "/emodb/kafka/lingerMs",
-                            "/" + UNIVERSE + "/emodb/kafka/bootstrapServers"
-                    )
+                    Stream.of("batchSize", "retries", "lingerMs", "bootstrapServers")
+                            .map(param -> basePath + param)
+                            .collect(Collectors.toList())
             );
 
             // Set configurations with fallback to defaults if not present
             // Sets the batch size for Kafka producer, which controls the amount of data to batch before sending.
-            batchSizeConfig = parameterValues.getOrDefault("/" + UNIVERSE + "/emodb/kafka/batchSize", "16384");
+            batchSizeConfig = parameterValues.getOrDefault(basePath+ "batchSize", "16384");
 
             // Sets the number of retry attempts for failed Kafka message sends.
-            retriesConfig = parameterValues.getOrDefault("/" + UNIVERSE + "/emodb/kafka/retries", "3");
+            retriesConfig = parameterValues.getOrDefault(basePath+ "retries", "3");
 
             // Sets the number of milliseconds a producer is willing to wait before sending a batch out
-            lingerMsConfig = parameterValues.getOrDefault("/" + UNIVERSE + "/emodb/kafka/lingerMs", "1");
+            lingerMsConfig = parameterValues.getOrDefault(basePath+"lingerMs", "1");
 
             // Configures the Kafka broker addresses for producer connections.
-            bootstrapServersConfig = parameterValues.getOrDefault("/" + UNIVERSE + "/emodb/kafka/bootstrapServers", DEFAULT_BOOTSTRAP_SERVERS);
+            bootstrapServersConfig = parameterValues.get(basePath+"bootstrapServers");
 
-            logger.info("Kafka configurations loaded successfully from SSM.");
-        } catch (AmazonServiceException e) {
-            logger.error("Failed to load configurations from SSM. Using default values.", e);
-            throw e;
+            // Log the kafka configurations loaded from SSM
+            logger.info("Kafka configurations loaded from SSM: batchSize={}, retries={}, lingerMs={}, bootstrapServers={}",
+                    batchSizeConfig, retriesConfig, lingerMsConfig,
+                    bootstrapServersConfig != null ? bootstrapServersConfig : "Not configured (null)");
         }
         catch (Exception e) {
             logger.error("Unexpected error occurred while loading configurations from SSM. Using default values.", e);

web/src/main/java/com/bazaarvoice/emodb/web/resources/queue/QueueResource1.java

@@ -127,7 +127,7 @@ public SuccessResponse sendBatch(@PathParam("queue") String queue, Collection<Ob
     @Path("{queue}/sendbatch1")
     @Consumes(MediaType.APPLICATION_JSON)
     @RequiresPermissions("queue|post|{queue}")
-    @Timed(name = "bv.emodb.queue.QueueResource1.sendBatch", absolute = true)
+    @Timed(name = "bv.emodb.queue.QueueResource1.sendBatch1", absolute = true)
     @ApiOperation (value = "Send a Batch.",
             notes = "Returns a SuccessResponse..",
             response = SuccessResponse.class