Kafka Extension

Signed-off-by: desmax74 <mdessi@redhat.com>
jboss-container-images · Mar 26, 2021 · 71d44ff · 71d44ff
1 parent 9a1965b
commit 71d44ff
Show file tree

Hide file tree

Showing 3 changed files with 135 additions and 1 deletion.
diff --git a/jboss-kie-kieserver/added/launch/jboss-kie-kieserver.sh b/jboss-kie-kieserver/added/launch/jboss-kie-kieserver.sh
@@ -26,6 +26,14 @@ function prepareEnv() {
     unset KIE_SERVER_DISABLE_KC_VERIFICATION
     unset KIE_SERVER_HOST
     unset KIE_SERVER_ID
+    unset KIE_SERVER_KAFKA_EXT_ACKS
+    unset KIE_SERVER_KAFKA_EXT_AUTOCREATE_TOPICS
+    unset KIE_SERVER_KAFKA_EXT_BOOTSTRAP_SERVERS
+    unset KIE_SERVER_KAFKA_EXT_CLIENT_ID
+    unset KIE_SERVER_KAFKA_EXT_ENABLED
+    unset KIE_SERVER_KAFKA_EXT_GROUP_ID
+    unset KIE_SERVER_KAFKA_EXT_MAX_BLOCK_MS
+    unset KIE_SERVER_KAFKA_EXT_TOPICS
     unset KIE_SERVER_LOCATION
     unset KIE_SERVER_MGMT_DISABLED
     unset KIE_SERVER_MODE
@@ -69,6 +77,7 @@ function configure() {
     configure_server_sync_deploy
     configure_drools
     configure_jbpm
+    configure_kafka
     configure_kie_server_mgmt
     configure_mode
     configure_metaspace
@@ -719,3 +728,36 @@ function configure_server_state() {
 function generate_random_id() {
     cat /dev/urandom | env LC_CTYPE=C tr -dc 'a-zA-Z0-9' | fold -w 16 | head -n 1
 }
+
+function configure_kafka(){
+  if [ "${KIE_SERVER_KAFKA_EXT_ENABLED^^}" = "TRUE" ]; then
+    if [ -n "${KIE_SERVER_KAFKA_EXT_BOOTSTRAP_SERVERS}" ];then
+      log_info "Kafka Extension enabled"
+      JBOSS_KIE_ARGS="${JBOSS_KIE_ARGS} -Dorg.kie.server.jbpm-kafka.ext.disabled=false"
+      JBOSS_KIE_ARGS="${JBOSS_KIE_ARGS} -Dorg.kie.server.jbpm-kafka.ext.bootstrap.servers=${KIE_SERVER_KAFKA_EXT_BOOTSTRAP_SERVERS}"
+      JBOSS_KIE_ARGS="${JBOSS_KIE_ARGS} -Dorg.kie.server.jbpm-kafka.ext.client.id=${KIE_SERVER_KAFKA_EXT_CLIENT_ID}"
+      JBOSS_KIE_ARGS="${JBOSS_KIE_ARGS} -Dorg.kie.server.jbpm-kafka.ext.group.id=${KIE_SERVER_KAFKA_EXT_GROUP_ID}"
+      JBOSS_KIE_ARGS="${JBOSS_KIE_ARGS} -Dorg.kie.server.jbpm-kafka.ext.acks=${KIE_SERVER_KAFKA_EXT_ACKS}"
+      JBOSS_KIE_ARGS="${JBOSS_KIE_ARGS} -Dorg.kie.server.jbpm-kafka.ext.max.block.ms=${KIE_SERVER_KAFKA_EXT_MAX_BLOCK_MS}"
+      JBOSS_KIE_ARGS="${JBOSS_KIE_ARGS} -Dorg.kie.server.jbpm-kafka.ext.allow.auto.create.topics=${KIE_SERVER_KAFKA_EXT_AUTOCREATE_TOPICS}"
+
+      IFS=',' read -a ks_topics <<< $KIE_SERVER_KAFKA_EXT_TOPICS
+      for topic in "${ks_topics[@]}"; do
+        IFS='=' read -a mapping <<< $topic
+        signal=${mapping[0]}
+        topic_name=${mapping[1]}
+          if [[ -n "$signal" && -n "$topic_name" ]]; then
+            JBOSS_KIE_ARGS="${JBOSS_KIE_ARGS} -Dorg.kie.server.jbpm-kafka.ext.topics.${topic}"
+          else
+              log_warning "mapping not configured, msg or topic name is empty. Value set [${signal}=${topic_name}]"
+          fi
+      done
+    else
+      JBOSS_KIE_ARGS="${JBOSS_KIE_ARGS} -Dorg.kie.server.jbpm-kafka.ext.disabled=true"
+      log_warning "Bootstrap servers not configured, kafka extension disabled"
+    fi
+  else
+    log_info "Kafka Extension disabled"
+    JBOSS_KIE_ARGS="${JBOSS_KIE_ARGS} -Dorg.kie.server.jbpm-kafka.ext.disabled=true"
+  fi
+}
diff --git a/jboss-kie-kieserver/tests/bats/jboss-kie-kieserver.bats b/jboss-kie-kieserver/tests/bats/jboss-kie-kieserver.bats
@@ -617,3 +617,52 @@ teardown() {
     echo "GC_MAX_METASPACE_SIZE=${GC_MAX_METASPACE_SIZE}"
     [[ "${GC_MAX_METASPACE_SIZE}" == "${KIE_SERVER_MAX_METASPACE_SIZE}" ]]
 }
+
+@test "Verify the Kafka extension" {
+    export KIE_SERVER_KAFKA_EXT_ENABLED="true"
+    export KIE_SERVER_KAFKA_EXT_BOOTSTRAP_SERVERS="localhost:9092"
+    export KIE_SERVER_KAFKA_EXT_CLIENT_ID="app"
+    export KIE_SERVER_KAFKA_EXT_GROUP_ID="jbpm-consumer"
+    export KIE_SERVER_KAFKA_EXT_ACKS="2"
+    export KIE_SERVER_KAFKA_EXT_MAX_BLOCK_MS="2000"
+    export KIE_SERVER_KAFKA_EXT_AUTOCREATE_TOPICS="true"
+    export KIE_SERVER_KAFKA_EXT_TOPICS="person=human,dog=animal,ant="
+
+    configure_kafka
+
+    local expected=" -Dorg.kie.server.jbpm-kafka.ext.disabled=false -Dorg.kie.server.jbpm-kafka.ext.bootstrap.servers=localhost:9092 -Dorg.kie.server.jbpm-kafka.ext.client.id=app -Dorg.kie.server.jbpm-kafka.ext.group.id=jbpm-consumer -Dorg.kie.server.jbpm-kafka.ext.acks=2 -Dorg.kie.server.jbpm-kafka.ext.max.block.ms=2000 -Dorg.kie.server.jbpm-kafka.ext.allow.auto.create.topics=true -Dorg.kie.server.jbpm-kafka.ext.topics.person=human -Dorg.kie.server.jbpm-kafka.ext.topics.dog=animal"
+    echo "  Result: ${JBOSS_KIE_ARGS}"
+
+    echo "Expected: ${expected}"
+    [[ "${JBOSS_KIE_ARGS}" == "${expected}" ]]
+}
+
+@test "Verify the Kafka extension disabled" {
+    export KIE_SERVER_KAFKA_EXT_ENABLED="false"
+
+    configure_kafka
+
+    local expected=" -Dorg.kie.server.jbpm-kafka.ext.disabled=true"
+    echo "  Result: ${JBOSS_KIE_ARGS}"
+
+    echo "Expected: ${expected}"
+    [[ "${JBOSS_KIE_ARGS}" == "${expected}" ]]
+}
+
+@test "Verify the Kafka extension without boootstrapservers" {
+    export KIE_SERVER_KAFKA_EXT_ENABLED="true"
+    export KIE_SERVER_KAFKA_EXT_CLIENT_ID="app"
+    export KIE_SERVER_KAFKA_EXT_GROUP_ID="jbpm-consumer"
+    export KIE_SERVER_KAFKA_EXT_ACKS="2"
+    export KIE_SERVER_KAFKA_EXT_MAX_BLOCK_MS="2000"
+    export KIE_SERVER_KAFKA_EXT_AUTOCREATE_TOPICS="true"
+    export KIE_SERVER_KAFKA_EXT_TOPICS="person=human,dog=animal,ant="
+
+    configure_kafka
+
+    local expected=" -Dorg.kie.server.jbpm-kafka.ext.disabled=true"
+    echo "  Result: ${JBOSS_KIE_ARGS}"
+
+    echo "Expected: ${expected}"
+    [[ "${JBOSS_KIE_ARGS}" == "${expected}" ]]
+}
diff --git a/tests/features/common/kie-kieserver-common.feature b/tests/features/common/kie-kieserver-common.feature
@@ -517,4 +517,47 @@ Feature: Kie Server common features
       | variable                            | value                          |
       | KIE_SERVER_CONTAINER_DEPLOYMENT     | test=org.package:mypackage:1.0 |
     Then container log should contain Using standard EnvVar KIE_SERVER_CONTAINER_DEPLOYMENT: test=org.package:mypackage:1.0
-     And container log should contain INFO Attempting to pull dependencies for kjar 0 with
+     And container log should contain INFO Attempting to pull dependencies for kjar 0 with
+
+  Scenario: Check if the Kafka integration is disabled
+    When container is started with env
+      | variable                       | value    |
+      | KIE_SERVER_KAFKA_EXT_ENABLED   | false    |
+    Then container log should contain -Dorg.kie.server.jbpm-kafka.ext.disabled=true
+
+  Scenario: Check if the Kafka integration is enabled
+    When container is started with env
+      | variable                               | value                         |
+      | KIE_SERVER_KAFKA_EXT_ENABLED           | true                          |
+      | KIE_SERVER_KAFKA_EXT_BOOTSTRAP_SERVERS | localhost:9092                |
+      | KIE_SERVER_KAFKA_EXT_CLIENT_ID         | app                           |
+      | KIE_SERVER_KAFKA_EXT_GROUP_ID          | jbpm-consumer                 |
+      | KIE_SERVER_KAFKA_EXT_ACKS              | 2                             |
+      | KIE_SERVER_KAFKA_EXT_MAX_BLOCK_MS      | 2000                          |
+      | KIE_SERVER_KAFKA_EXT_AUTOCREATE_TOPICS | true                          |
+      | KIE_SERVER_KAFKA_EXT_TOPICS            | person=human,dog=animal,ant=  |
+      | SCRIPT_DEBUG                           | true                          |
+    Then container log should contain -Dorg.kie.server.jbpm-kafka.ext.disabled=false
+    And container log should contain -Dorg.kie.server.jbpm-kafka.ext.bootstrap.servers=localhost:9092
+    And container log should contain -Dorg.kie.server.jbpm-kafka.ext.client.id=app
+    And container log should contain -Dorg.kie.server.jbpm-kafka.ext.group.id=jbpm-consumer
+    And container log should contain -Dorg.kie.server.jbpm-kafka.ext.acks=2
+    And container log should contain -Dorg.kie.server.jbpm-kafka.ext.max.block.ms=2000
+    And container log should contain -Dorg.kie.server.jbpm-kafka.ext.allow.auto.create.topics=true
+    And container log should contain -Dorg.kie.server.jbpm-kafka.ext.topics.person=human
+    And container log should contain -Dorg.kie.server.jbpm-kafka.ext.topics.dog=animal
+    And container log should contain mapping not configured, msg or topic name is empty. Value set [ant=]
+
+  Scenario: Check if the Kafka integration is enabled without bootstrapservers
+    When container is started with env
+      | variable                               | value                         |
+      | KIE_SERVER_KAFKA_EXT_ENABLED           | true                          |
+      | KIE_SERVER_KAFKA_EXT_CLIENT_ID         | app                           |
+      | KIE_SERVER_KAFKA_EXT_GROUP_ID          | jbpm-consumer                 |
+      | KIE_SERVER_KAFKA_EXT_ACKS              | 2                             |
+      | KIE_SERVER_KAFKA_EXT_MAX_BLOCK_MS      | 2000                          |
+      | KIE_SERVER_KAFKA_EXT_AUTOCREATE_TOPICS | true                          |
+      | KIE_SERVER_KAFKA_EXT_TOPICS            | person=human,dog=animal,ant=  |
+      | SCRIPT_DEBUG                           | true                          |
+    Then container log should contain -Dorg.kie.server.jbpm-kafka.ext.disabled=true
+    And container log should contain Bootstrap servers not configured, kafka extension disabled