OP-23081: Corrected kafka consumer group and endpoint

OpsMx · Jan 31, 2025 · 027ff8d · 027ff8d
2 parents 190b2dd + 7b6ac63
commit 027ff8d
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 9 deletions.
diff --git a/gate-web/gate-web.gradle b/gate-web/gate-web.gradle
@@ -68,6 +68,7 @@ dependencies {
     exclude group: 'com.rabbitmq', module: 'amqp-client'
   }
   implementation('org.apache.camel:camel-kafka:3.14.1')
+  implementation('software.amazon.msk:aws-msk-iam-auth:2.2.0')
   implementation ("com.rabbitmq:amqp-client:5.18.0") {
     force(true)
   }

diff --git a/gate-web/src/main/java/com/netflix/spinnaker/gate/config/KafkaConfig.java b/gate-web/src/main/java/com/netflix/spinnaker/gate/config/KafkaConfig.java
@@ -1,7 +1,5 @@
 package com.netflix.spinnaker.gate.config;
 
-import static com.opsmx.spinnaker.gate.constant.CamelEndpointConstant.directUserActivity;
-
 import java.util.Objects;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -17,13 +15,11 @@ public class KafkaConfig implements CamelRouteConfig {
 
   @Autowired private KafkaProperties kafkaProperties;
 
-  private static final String groupId = "oes-sapor-consumer-group";
+  private static final String groupId = "oes-gate-consumer-group";
 
   @Override
   public String getUserActivityQueueEndPoint() {
-    return "kafka:"
-        + directUserActivity
-        + "?brokers="
+    return "kafka:isd-userLoginDetails?brokers="
         + kafkaProperties.getBootstrapAddress()
         + "&groupid="
         + groupId
@@ -71,7 +67,7 @@ private String getSecurityString() {
               + "&saslMechanism="
               + kafkaProperties.getSasl().getMechanism()
               + "&saslJaasConfig=software.amazon.msk.auth.iam.IAMLoginModule required;"
-              + "&saslClientCallbackHandlerClass=software.amazon.msk.auth.iam.IAMClientCallbackHandler";
+              + "&additionalProperties.sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler";
     }
 
     if (kafkaProperties.getSecurity() != null
@@ -83,8 +79,10 @@ private String getSecurityString() {
               + "&saslMechanism="
               + kafkaProperties.getSasl().getMechanism()
               + "&saslJaasConfig=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;"
-              + "&saslClientCallbackHandlerClass=software.amazon.msk.auth.iam.IAMOAuthBearerLoginCallbackHandler"
-              + "&saslClientCallbackHandlerClass=software.amazon.msk.auth.iam.IAMOAuthBearerLoginCallbackHandler";
+              + "&additionalProperties.sasl.login.callback.handler.class=software.amazon.msk.auth.iam.IAMOAuthBearerLoginCallbackHandler"
+              + "&additionalProperties.sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMOAuthBearerLoginCallbackHandler";
+      System.setProperty("aws.accessKeyId", kafkaProperties.getAccessKeyId());
+      System.setProperty("aws.secretKey", kafkaProperties.getSecretKey());
     }
     return securityString;
   }

diff --git a/gate-web/src/main/java/com/netflix/spinnaker/gate/config/KafkaProperties.java b/gate-web/src/main/java/com/netflix/spinnaker/gate/config/KafkaProperties.java
@@ -15,6 +15,8 @@ public class KafkaProperties {
   private Security security;
   private Sasl sasl;
   private Ssl ssl;
+  private String accessKeyId;
+  private String secretKey;
 
   @Data
   @Configuration