#2530 Enable SASL for bie kafka

app/docker-compose.yml

@@ -128,5 +128,7 @@ services:
       BIE_KAFKA_KEYSTORE_PASSWORD: ${BIE_KAFKA_KEYSTORE_PASSWORD}
       BIE_KAFKA_TRUSTSTORE_INBASE64: ${BIE_KAFKA_TRUSTSTORE_INBASE64}
       BIE_KAFKA_TRUSTSTORE_PASSWORD: ${BIE_KAFKA_TRUSTSTORE_PASSWORD}
+      BIE_KAFKA_RBAC_USERNAME: ${BIE_KAFKA_RBAC_USERNAME}
+      BIE_KAFKA_RBAC_PASSWORD: ${BIE_KAFKA_RBAC_PASSWORD}
     networks:
       - vro_intranet

scripts/setenv.sh

@@ -253,3 +253,5 @@ exportSecretIfUnset BIP_KEYSTORE
 exportSecretIfUnset BIP_PASSWORD
 exportSecretIfUnset BIP_CLAIM_URL
 exportSecretIfUnset BIP_EVIDENCE_URL
+export BIE_KAFKA_RBAC_USERNAME=USERNAME
+export BIE_KAFKA_RBAC_PASSWORD=PASSWORD

shared/lib-bie-kafka/src/main/java/gov/va/vro/model/biekafka/BieMessagePayload.java

@@ -34,60 +34,60 @@ public class BieMessagePayload {
 
   // populated from kafka topic payload
   @TargetEvents({
-    "CONTENTION_BIE_CONTENTION_ASSOCIATED_TO_CLAIM_V02",
-    "CONTENTION_BIE_CONTENTION_CLASSIFIED_V02",
-    "CONTENTION_BIE_CONTENTION_UPDATED_V02"
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_ASSOCIATED_TO_CLAIM_V02",
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_CLASSIFIED_V02",
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_UPDATED_V02"
   })
   private String benefitClaimTypeCode;
 
   @TargetEvents({
-    "CONTENTION_BIE_CONTENTION_ASSOCIATED_TO_CLAIM_V02",
-    "CONTENTION_BIE_CONTENTION_CLASSIFIED_V02",
-    "CONTENTION_BIE_CONTENTION_UPDATED_V02"
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_ASSOCIATED_TO_CLAIM_V02",
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_CLASSIFIED_V02",
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_UPDATED_V02"
   })
   private String actorStation;
 
   @TargetEvents({
-    "CONTENTION_BIE_CONTENTION_ASSOCIATED_TO_CLAIM_V02",
-    "CONTENTION_BIE_CONTENTION_CLASSIFIED_V02",
-    "CONTENTION_BIE_CONTENTION_UPDATED_V02"
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_ASSOCIATED_TO_CLAIM_V02",
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_CLASSIFIED_V02",
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_UPDATED_V02"
   })
   private String details;
 
   @TargetEvents({
-    "CONTENTION_BIE_CONTENTION_ASSOCIATED_TO_CLAIM_V02",
-    "CONTENTION_BIE_CONTENTION_CLASSIFIED_V02",
-    "CONTENTION_BIE_CONTENTION_UPDATED_V02"
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_ASSOCIATED_TO_CLAIM_V02",
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_CLASSIFIED_V02",
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_UPDATED_V02"
   })
   private Long veteranParticipantId;
 
   @TargetEvents({
-    "CONTENTION_BIE_CONTENTION_ASSOCIATED_TO_CLAIM_V02",
-    "CONTENTION_BIE_CONTENTION_CLASSIFIED_V02",
-    "CONTENTION_BIE_CONTENTION_UPDATED_V02"
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_ASSOCIATED_TO_CLAIM_V02",
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_CLASSIFIED_V02",
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_UPDATED_V02"
   })
   private String contentionClassificationName;
 
   @TargetEvents({
-    "CONTENTION_BIE_CONTENTION_ASSOCIATED_TO_CLAIM_V02",
-    "CONTENTION_BIE_CONTENTION_CLASSIFIED_V02",
-    "CONTENTION_BIE_CONTENTION_UPDATED_V02"
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_ASSOCIATED_TO_CLAIM_V02",
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_CLASSIFIED_V02",
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_UPDATED_V02"
   })
   private String diagnosticTypeCode;
 
-  @TargetEvents({"CONTENTION_BIE_CONTENTION_UPDATED_V02"})
+  @TargetEvents({"BIA_SERVICES_BIE_CATALOG_CONTENTION_UPDATED_V02"})
   private String journalStatusTypeCode;
 
   @TargetEvents({
-    "CONTENTION_BIE_CONTENTION_ASSOCIATED_TO_CLAIM_V02",
-    "CONTENTION_BIE_CONTENTION_CLASSIFIED_V02",
-    "CONTENTION_BIE_CONTENTION_UPDATED_V02"
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_ASSOCIATED_TO_CLAIM_V02",
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_CLASSIFIED_V02",
+    "BIA_SERVICES_BIE_CATALOG_CONTENTION_UPDATED_V02"
   })
   private Long dateAdded;
 
-  @TargetEvents({"CONTENTION_BIE_CONTENTION_UPDATED_V02"})
+  @TargetEvents({"BIA_SERVICES_BIE_CATALOG_CONTENTION_UPDATED_V02"})
   private Long dateCompleted;
 
-  @TargetEvents({"CONTENTION_BIE_CONTENTION_UPDATED_V02"})
+  @TargetEvents({"BIA_SERVICES_BIE_CATALOG_CONTENTION_UPDATED_V02"})
   private Long dateUpdated;
 }

shared/lib-bie-kafka/src/main/java/gov/va/vro/model/biekafka/ContentionEvent.java

@@ -3,11 +3,11 @@
 import java.util.Arrays;
 
 public enum ContentionEvent {
-  CONTENTION_ASSOCIATED_TO_CLAIM("CONTENTION_BIE_CONTENTION_ASSOCIATED_TO_CLAIM_V02"),
-  CONTENTION_UPDATED("CONTENTION_BIE_CONTENTION_UPDATED_V02"),
-  CONTENTION_CLASSIFIED("CONTENTION_BIE_CONTENTION_CLASSIFIED_V02"),
-  CONTENTION_COMPLETED("CONTENTION_BIE_CONTENTION_COMPLETED_V02"),
-  CONTENTION_DELETED("CONTENTION_BIE_CONTENTION_DELETED_V02");
+  CONTENTION_ASSOCIATED_TO_CLAIM("BIA_SERVICES_BIE_CATALOG_CONTENTION_ASSOCIATED_TO_CLAIM_V02"),
+  CONTENTION_UPDATED("BIA_SERVICES_BIE_CATALOG_CONTENTION_UPDATED_V02"),
+  CONTENTION_CLASSIFIED("BIA_SERVICES_BIE_CATALOG_CONTENTION_CLASSIFIED_V02"),
+  CONTENTION_COMPLETED("BIA_SERVICES_BIE_CATALOG_CONTENTION_COMPLETED_V02"),
+  CONTENTION_DELETED("BIA_SERVICES_BIE_CATALOG_CONTENTION_DELETED_V02");
 
   private final String topicName;
 
@@ -20,13 +20,13 @@ public String getTopicName() {
   }
 
   public static ContentionEvent mapTopicToEvent(String topic) {
-    // remove first word prefix from topic seperated by _
-    String noPrefixTopic = topic.substring(topic.indexOf("_") + 1);
+    String subString = "CATALOG_.*?_CONTENTION";
+    String noSubStringTopic = topic.replaceAll(subString, "CATALOG_CONTENTION");
 
     return Arrays.stream(ContentionEvent.values())
-        .filter(event -> event.getTopicName().equals(noPrefixTopic))
+        .filter(event -> event.getTopicName().equals(noSubStringTopic))
         .findFirst()
-        .orElseThrow(() -> new IllegalArgumentException("Unrecognized topic: " + noPrefixTopic));
+        .orElseThrow(() -> new IllegalArgumentException("Unrecognized topic: " + noSubStringTopic));
   }
 
   public static String rabbitMqExchangeName(String topic) {

svc-bie-kafka/docker-entryprep.sh

@@ -6,20 +6,12 @@ getEnvVarValue(){
 }
 
 for ENV_VAR in \
-  BIE_KAFKA_KEYSTORE_INBASE64 BIE_KAFKA_KEYSTORE_PASSWORD \
   BIE_KAFKA_TRUSTSTORE_INBASE64 BIE_KAFKA_TRUSTSTORE_PASSWORD; do
   if [ "$(getEnvVarValue "$ENV_VAR")" = "" ]; then
     >&2 echo "ERROR: Missing expected environment variable: $ENV_VAR"
   fi
 done
 
-export KEYSTORE_FILE="$PWD/keystore.p12"
-echo "$BIE_KAFKA_KEYSTORE_INBASE64" | base64 -d > "$KEYSTORE_FILE"
-echo -e "\nVerifying keystore ($KEYSTORE_FILE) and its password..."
-if ! keytool -list -v -keystore "$KEYSTORE_FILE" -storepass "$BIE_KAFKA_KEYSTORE_PASSWORD" | grep "Alias name:"; then
-  >&2 echo "ERROR: with keystore"
-fi
-
 export TRUSTSTORE_FILE="$PWD/truststore.p12"
 echo "$BIE_KAFKA_TRUSTSTORE_INBASE64" | base64 -d > "$TRUSTSTORE_FILE"
 echo -e "\nVerifying truststore ($TRUSTSTORE_FILE) and its password..."

svc-bie-kafka/src/main/java/gov/va/vro/services/bie/config/BieProperties.java

@@ -13,11 +13,15 @@
 @Setter
 public class BieProperties {
 
-  @Getter String kakfaTopicPrefix;
+  @Getter String kafkaTopicInfix;
 
   public String[] topicNames() {
     return Arrays.stream(ContentionEvent.values())
-        .map(contention -> kakfaTopicPrefix + contention.getTopicName())
+        .map(
+            contention -> {
+              String subString = "CATALOG_" + kafkaTopicInfix + "_CONTENTION";
+              return contention.getTopicName().replaceAll("CATALOG_CONTENTION", subString);
+            })
         .toArray(String[]::new);
   }
 }

svc-bie-kafka/src/main/resources/application-dev.yaml

@@ -1,24 +1,32 @@
 spring:
   kafka:
-    bootstrap-servers: "${BIE_KAFKA_PLACEHOLDERS_BROKERS:kafka.dev.bip.va.gov:443}"
+    bootstrap-servers: "${BIE_KAFKA_PLACEHOLDERS_BROKERS:bip-kafka.dev.bip.va.gov:443}"
     properties:
-      schema.registry.url: "${BIE_KAFKA_PLACEHOLDERS_BROKERS:https://schemaregistry.dev.bip.va.gov:443}"
-      schema.registry.ssl.protocol: SSL
-      schema.registry.ssl.keystore.location: "${KEYSTORE_FILE}"
-      schema.registry.ssl.keystore.password: "${BIE_KAFKA_KEYSTORE_PASSWORD}"
-      schema.registry.ssl.keystore.type: "PKCS12"
-      schema.registry.ssl.truststore.location: "${TRUSTSTORE_FILE}"
-      schema.registry.ssl.truststore.password: "${BIE_KAFKA_TRUSTSTORE_PASSWORD}"
-      schema.registry.ssl.truststore.type: "PKCS12"
+      basic.auth.credentials.source: USER_INFO
+      basic.auth.user.info: "${BIE_KAFKA_RBAC_USERNAME}:${BIE_KAFKA_RBAC_PASSWORD}"
+      schema.registry:
+        basic.auth.credentials.source: USER_INFO
+        basic.auth.user.info: "${BIE_KAFKA_RBAC_USERNAME}:${BIE_KAFKA_RBAC_PASSWORD}"
+        url: "${BIE_KAFKA_PLACEHOLDERS_BROKERS:https://bip-schemaregistry.dev.bip.va.gov}"
+        ssl:
+          truststore.location: "${TRUSTSTORE_FILE}"
+          truststore.password: "${BIE_KAFKA_TRUSTSTORE_PASSWORD}"
+          truststore.type: "PKCS12"
+      security.protocol: SASL_SSL
+      sasl:
+        mechanism: PLAIN
+        jaas:
+          config: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"${BIE_KAFKA_RBAC_USERNAME}\" password=\"${BIE_KAFKA_RBAC_PASSWORD}\";"
     consumer:
-      group-id: "${BIE_KAFKA_PLACEHOLDERS_GROUP_ID:vro-bie-tst-vro-1}"
+      group-id: "${BIE_KAFKA_PLACEHOLDERS_GROUP_ID:EXT_VRO_TST}"
       key-deserializer: "org.apache.kafka.common.serialization.StringDeserializer"
       value-deserializer: "io.confluent.kafka.serializers.KafkaAvroDeserializer"
-      properties:
-        security.protocol: SSL
-        ssl.keystore.type: PKCS12
-        ssl.keystore.location: "${KEYSTORE_FILE}"
-        ssl.keystore.password: "${BIE_KAFKA_KEYSTORE_PASSWORD}"
-        ssl.truststore.type: PKCS12
-        ssl.truststore.location: "${TRUSTSTORE_FILE}"
-        ssl.truststore.password: "${BIE_KAFKA_TRUSTSTORE_PASSWORD}"
+      security:
+        protocol: SASL_SSL
+      ssl:
+        trust-store-location: "file:${TRUSTSTORE_FILE}"
+        trust-store-password: "${BIE_KAFKA_TRUSTSTORE_PASSWORD}"
+        trust-store-type: "PKCS12"
+
+bie:
+  kafka-topic-infix: "TST"

svc-bie-kafka/src/main/resources/application-integration-test.yaml

@@ -1,4 +1,3 @@
-
 spring:
   kafka:
     # To prevent java.net.UnknownHostException mock-bie-kafka, use mock Kafka's 'EXTERNAL' port 9094

svc-bie-kafka/src/main/resources/application-qa.yaml

@@ -1,35 +1,32 @@
 spring:
   kafka:
-    bootstrap-servers: "${BIE_KAFKA_PLACEHOLDERS_BROKERS:kafka.stage.bip.va.gov:443}"
+    bootstrap-servers: "${BIE_KAFKA_PLACEHOLDERS_BROKERS:bip-kafka.qa.bip.va.gov:443}"
     properties:
-      schema:
-        registry:
-          url: "${BIE_KAFKA_PLACEHOLDERS_BROKERS:https://schemaregistry.stage.bip.va.gov:443}"
-          ssl:
-            protocol: SSL
-            keystore:
-              type: PKCS12
-              location: "${KEYSTORE_FILE}"
-              password: "${BIE_KAFKA_KEYSTORE_PASSWORD}"
-            truststore:
-              type: PKCS12
-              location: "${TRUSTSTORE_FILE}"
-              password: "${BIE_KAFKA_TRUSTSTORE_PASSWORD}"
+      basic.auth.credentials.source: USER_INFO
+      basic.auth.user.info: "${BIE_KAFKA_RBAC_USERNAME}:${BIE_KAFKA_RBAC_PASSWORD}"
+      schema.registry:
+        basic.auth.credentials.source: USER_INFO
+        basic.auth.user.info: "${BIE_KAFKA_RBAC_USERNAME}:${BIE_KAFKA_RBAC_PASSWORD}"
+        url: "${BIE_KAFKA_PLACEHOLDERS_BROKERS:https://bip-schemaregistry.dev.bip.va.gov}"
+        ssl:
+          truststore.location: "${TRUSTSTORE_FILE}"
+          truststore.password: "${BIE_KAFKA_TRUSTSTORE_PASSWORD}"
+          truststore.type: "PKCS12"
+      security.protocol: SASL_SSL
+      sasl:
+        mechanism: PLAIN
+        jaas:
+          config: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"${BIE_KAFKA_RBAC_USERNAME}\" password=\"${BIE_KAFKA_RBAC_PASSWORD}\";"
     consumer:
-      group-id: "${BIE_KAFKA_PLACEHOLDERS_GROUP_ID:vro-bie-ivv-vro}"
+      group-id: "${BIE_KAFKA_PLACEHOLDERS_GROUP_ID:EXT_VRO_QA}"
       key-deserializer: "org.apache.kafka.common.serialization.StringDeserializer"
       value-deserializer: "io.confluent.kafka.serializers.KafkaAvroDeserializer"
-      properties:
-        security.protocol: SSL
-        ssl:
-          keystore:
-            type: PKCS12
-            location: "${KEYSTORE_FILE}"
-            password: "${BIE_KAFKA_KEYSTORE_PASSWORD}"
-          truststore:
-            type: PKCS12
-            location: "${TRUSTSTORE_FILE}"
-            password: "${BIE_KAFKA_TRUSTSTORE_PASSWORD}"
+      security:
+        protocol: SASL_SSL
+      ssl:
+        trust-store-location: "file:${TRUSTSTORE_FILE}"
+        trust-store-password: "${BIE_KAFKA_TRUSTSTORE_PASSWORD}"
+        trust-store-type: "PKCS12"
 
 bie:
-  kakfa-topic-prefix: "IVV_"
+  kafka-topic-infix: "QA"

svc-bie-kafka/src/main/resources/application-sandbox.yaml

@@ -1,35 +1,32 @@
 spring:
   kafka:
-    bootstrap-servers: "${BIE_KAFKA_PLACEHOLDERS_BROKERS:kafka.stage.bip.va.gov:443}"
+    bootstrap-servers: "${BIE_KAFKA_PLACEHOLDERS_BROKERS:bip-kafka.stage.bip.va.gov:443}"
     properties:
-      schema:
-        registry:
-          url: "${BIE_KAFKA_PLACEHOLDERS_BROKERS:https://schemaregistry.stage.bip.va.gov:443}"
-          ssl:
-            protocol: SSL
-            keystore:
-              type: PKCS12
-              location: "${KEYSTORE_FILE}"
-              password: "${BIE_KAFKA_KEYSTORE_PASSWORD}"
-            truststore:
-              type: PKCS12
-              location: "${TRUSTSTORE_FILE}"
-              password: "${BIE_KAFKA_TRUSTSTORE_PASSWORD}"
+      basic.auth.credentials.source: USER_INFO
+      basic.auth.user.info: "${BIE_KAFKA_RBAC_USERNAME}:${BIE_KAFKA_RBAC_PASSWORD}"
+      schema.registry:
+        basic.auth.credentials.source: USER_INFO
+        basic.auth.user.info: "${BIE_KAFKA_RBAC_USERNAME}:${BIE_KAFKA_RBAC_PASSWORD}"
+        url: "${BIE_KAFKA_PLACEHOLDERS_BROKERS:bip-schemaregistry.stage.bip.va.go}"
+        ssl:
+          truststore.location: "${TRUSTSTORE_FILE}"
+          truststore.password: "${BIE_KAFKA_TRUSTSTORE_PASSWORD}"
+          truststore.type: "PKCS12"
+      security.protocol: SASL_SSL
+      sasl:
+        mechanism: PLAIN
+        jaas:
+          config: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"${BIE_KAFKA_RBAC_USERNAME}\" password=\"${BIE_KAFKA_RBAC_PASSWORD}\";"
     consumer:
-      group-id: "${BIE_KAFKA_PLACEHOLDERS_GROUP_ID:vro-bie-uat-vro}"
+      group-id: "${BIE_KAFKA_PLACEHOLDERS_GROUP_ID:EXT_VRO_IVV}"
       key-deserializer: "org.apache.kafka.common.serialization.StringDeserializer"
       value-deserializer: "io.confluent.kafka.serializers.KafkaAvroDeserializer"
-      properties:
-        security.protocol: SSL
-        ssl:
-          keystore:
-            type: PKCS12
-            location: "${KEYSTORE_FILE}"
-            password: "${BIE_KAFKA_KEYSTORE_PASSWORD}"
-          truststore:
-            type: PKCS12
-            location: "${TRUSTSTORE_FILE}"
-            password: "${BIE_KAFKA_TRUSTSTORE_PASSWORD}"
+    security:
+      protocol: SASL_SSL
+    ssl:
+      trust-store-location: "file:${TRUSTSTORE_FILE}"
+      trust-store-password: "${BIE_KAFKA_TRUSTSTORE_PASSWORD}"
+      trust-store-type: "PKCS12"
 
 bie:
-  kakfa-topic-prefix: "UAT_"
+  kafka-topic-infix: "IVV"

svc-bie-kafka/src/main/resources/application.yaml

@@ -33,7 +33,7 @@ spring:
 
 ## Specify bie properties
 bie:
-  kakfa-topic-prefix: "TST_"
+  kafka-topic-infix: "TST"
 
 ## Actuator for health check, liveness, and readiness
 management:

svc-bie-kafka/src/test/java/gov/va/vro/services/bie/config/ContentionEventTest.java

@@ -13,11 +13,11 @@ public class ContentionEventTest {
 
   @ParameterizedTest
   @CsvSource({
-    "TST_CONTENTION_BIE_CONTENTION_ASSOCIATED_TO_CLAIM_V02, CONTENTION_ASSOCIATED_TO_CLAIM",
-    "TST_CONTENTION_BIE_CONTENTION_UPDATED_V02, CONTENTION_UPDATED",
-    "TST_CONTENTION_BIE_CONTENTION_CLASSIFIED_V02, CONTENTION_CLASSIFIED",
-    "TST_CONTENTION_BIE_CONTENTION_COMPLETED_V02, CONTENTION_COMPLETED",
-    "TST_CONTENTION_BIE_CONTENTION_DELETED_V02, CONTENTION_DELETED"
+    "BIA_SERVICES_BIE_CATALOG_TST_CONTENTION_ASSOCIATED_TO_CLAIM_V02, CONTENTION_ASSOCIATED_TO_CLAIM",
+    "BIA_SERVICES_BIE_CATALOG_TST_CONTENTION_UPDATED_V02, CONTENTION_UPDATED",
+    "BIA_SERVICES_BIE_CATALOG_TST_CONTENTION_CLASSIFIED_V02, CONTENTION_CLASSIFIED",
+    "BIA_SERVICES_BIE_CATALOG_TST_CONTENTION_COMPLETED_V02, CONTENTION_COMPLETED",
+    "BIA_SERVICES_BIE_CATALOG_TST_CONTENTION_DELETED_V02, CONTENTION_DELETED"
   })
   public void testMapTopicToEvent_validTopics(String inputTopic, ContentionEvent expectedEvent) {
     assertEquals(expectedEvent, mapTopicToEvent(inputTopic));
@@ -35,11 +35,11 @@ public void testMapTopicToEvent_unrecognizedTopic() {
 
   @ParameterizedTest
   @CsvSource({
-    "TST_CONTENTION_BIE_CONTENTION_ASSOCIATED_TO_CLAIM_V02, bie-events-contention-associated-to-claim",
-    "TST_CONTENTION_BIE_CONTENTION_UPDATED_V02, bie-events-contention-updated",
-    "TST_CONTENTION_BIE_CONTENTION_CLASSIFIED_V02, bie-events-contention-classified",
-    "TST_CONTENTION_BIE_CONTENTION_COMPLETED_V02, bie-events-contention-completed",
-    "TST_CONTENTION_BIE_CONTENTION_DELETED_V02, bie-events-contention-deleted"
+    "BIA_SERVICES_BIE_CATALOG_TST_CONTENTION_ASSOCIATED_TO_CLAIM_V02, bie-events-contention-associated-to-claim",
+    "BIA_SERVICES_BIE_CATALOG_TST_CONTENTION_UPDATED_V02, bie-events-contention-updated",
+    "BIA_SERVICES_BIE_CATALOG_TST_CONTENTION_CLASSIFIED_V02, bie-events-contention-classified",
+    "BIA_SERVICES_BIE_CATALOG_TST_CONTENTION_COMPLETED_V02, bie-events-contention-completed",
+    "BIA_SERVICES_BIE_CATALOG_TST_CONTENTION_DELETED_V02, bie-events-contention-deleted"
   })
   public void testGenerateRabbitMQChannelName_channelNames(String inputTopic, String bieChannel) {
     assertEquals(bieChannel, ContentionEvent.rabbitMqExchangeName(inputTopic));