Improve SSL Expiry Metrics and Switch to Round-Robin Partition Assigner

src/main/java/hlf/java/rest/client/config/BaseKafkaConfig.java

@@ -4,7 +4,10 @@
 import io.micrometer.core.instrument.Gauge;
 import io.micrometer.core.instrument.MeterRegistry;
 import java.sql.Timestamp;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
 import java.time.Instant;
+import java.util.Date;
 import java.util.Map;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.BooleanUtils;
@@ -57,18 +60,24 @@ protected void configureSSLProperties(
             SSLAuthFilesHelper.getExpiryTimestampForKeyStore(
                 sslProperties.getSslTruststoreLocation(), sslProperties.getSslTruststorePassword());
 
+        DateFormat formatter = new SimpleDateFormat("dd/MM/yyyy");
+
         String guagePrefix =
             getConfigType().equals(ConfigType.CONSUMER) ? "consumer." : "producer.";
 
         Gauge.builder(
                 guagePrefix + topicName + ".keystore.expiryTs",
                 keyStoreCertExpiryTimestamp::getTime)
+            .tag("topic-name", topicName)
+            .tag("expiry-date", formatter.format(new Date(keyStoreCertExpiryTimestamp.getTime())))
             .strongReference(true)
             .register(sslMetricsRegistry);
 
         Gauge.builder(
                 guagePrefix + topicName + ".truststore.expiryTs",
                 trustStoreCertExpiryTimestamp::getTime)
+            .tag("topic-name", topicName)
+            .tag("expiry-date", formatter.format(new Date(trustStoreCertExpiryTimestamp.getTime())))
             .strongReference(true)
             .register(sslMetricsRegistry);
 
@@ -81,13 +90,15 @@ protected void configureSSLProperties(
                 guagePrefix + topicName + ".keystore.hasExpired",
                 hasKeyStoreCertExpired,
                 BooleanUtils::toInteger)
+            .tag("topic-name", topicName)
             .strongReference(true)
             .register(sslMetricsRegistry);
 
         Gauge.builder(
                 guagePrefix + topicName + ".truststore.hasExpired",
                 hasTrustStoreCertExpired,
                 BooleanUtils::toInteger)
+            .tag("topic-name", topicName)
             .strongReference(true)
             .register(sslMetricsRegistry);
 

src/main/java/hlf/java/rest/client/config/KafkaConsumerConfig.java

@@ -43,6 +43,11 @@ public DefaultKafkaConsumerFactory<String, String> consumerFactory(
     props.put(
         ConsumerConfig.MAX_POLL_RECORDS_CONFIG, FabricClientConstants.KAFKA_INTG_MAX_POLL_RECORDS);
 
+    // Distribute available partitions evenly across all consumers (or consumer threads)
+    props.put(
+        ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
+        FabricClientConstants.ROUND_ROBIN_CONSUMER_PARTITION_ASSIGNEMENT_STRATEGY);
+
     // Azure event-hub config
     configureSaslProperties(props, kafkaConsumerProperties.getSaslJaasConfig());
 

src/main/java/hlf/java/rest/client/config/SSLAuthFilesHelper.java

@@ -12,10 +12,14 @@
 import java.security.cert.CertificateException;
 import java.security.cert.X509Certificate;
 import java.sql.Timestamp;
+import java.util.ArrayList;
 import java.util.Base64;
+import java.util.Collections;
 import java.util.Enumeration;
+import java.util.List;
 import lombok.experimental.UtilityClass;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.util.CollectionUtils;
 
 @Slf4j
 @UtilityClass
@@ -57,18 +61,25 @@ Timestamp getExpiryTimestampForKeyStore(String keyStorePath, String keyStorePass
 
     KeyStore keyStore = loadKeyStore(keyStorePath, keyStorePassword);
 
+    List<Timestamp> certExpiryTimestamps = new ArrayList<>();
+
     Enumeration<String> aliases = keyStore.aliases();
     while (aliases.hasMoreElements()) {
       String alias = aliases.nextElement();
       Certificate cert = keyStore.getCertificate(alias);
       if (cert instanceof X509Certificate) {
         X509Certificate x509Cert = (X509Certificate) cert;
-        return new Timestamp(x509Cert.getNotAfter().getTime());
+        certExpiryTimestamps.add(new Timestamp(x509Cert.getNotAfter().getTime()));
       }
     }
 
-    throw new CertificateException(
-        "Couldn't extract an instance of X509Certificate for fetching expiry details");
+    if (CollectionUtils.isEmpty(certExpiryTimestamps)) {
+      throw new CertificateException(
+          "Couldn't extract an instance of X509Certificate for fetching expiry details");
+    }
+
+    // Return the earliest (minimum) timestamp from the list
+    return Collections.min(certExpiryTimestamps);
   }
 
   private static KeyStore loadKeyStore(String path, String password)

src/main/java/hlf/java/rest/client/util/FabricClientConstants.java

@@ -80,4 +80,6 @@ public class FabricClientConstants {
   public static final String JSON_PATH_CERTIFICATE = "$..certificate";
 
   public static final String VALUE_TAG_CAPABILITIES = "Capabilities";
+  public static final String ROUND_ROBIN_CONSUMER_PARTITION_ASSIGNEMENT_STRATEGY =
+      "org.apache.kafka.clients.consumer.RoundRobinAssignor";
 }