fix(native): register javax.security.auth.Subject for GraalVM reflection (#726)

SASL authentication fails in the native binary because the Kafka client's
CompositeStrategy uses reflection to find Subject.current() and Subject.callAs(),
which are not registered in the GraalVM native image metadata.

Add reflection entries for javax.security.auth.Subject, PlainSaslClientFactory,
and PlainSaslClientProvider to the Kafka provider's reachability-metadata.json.

Also add E2E tests for SASL_PLAINTEXT authentication: a SASL-enabled Kafka broker
via docker-compose, HOCON config, topic resources, and test scenarios integrated
into the existing run-tests.sh suite. A standalone run-sasl-tests.sh is also
provided for focused SASL testing.

Author: fhussonnois

e2e/docker-compose-e2e-sasl.yml

@@ -0,0 +1,49 @@
+#
+# SPDX-License-Identifier: Apache-2.0
+# Copyright (c) The original authors
+#
+# Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+#
+# Docker Compose for Jikkou e2e SASL tests.
+# Minimal setup: Kafka (KRaft) with SASL_PLAINTEXT authentication.
+services:
+  kafka-sasl:
+    image: "confluentinc/cp-kafka:7.5.0"
+    ports:
+      - "9093:9093"
+    container_name: e2e-kafka-sasl
+    environment:
+      KAFKA_NODE_ID: 101
+      CLUSTER_ID: 'xtzWWN4bTjitpL3kfd9s5g'
+      KAFKA_CONTROLLER_QUORUM_VOTERS: '101@kafka-sasl:29094'
+      KAFKA_PROCESS_ROLES: 'broker,controller'
+      KAFKA_ADVERTISED_LISTENERS: 'SASL_PLAINTEXT://kafka-sasl:29093,SASL_PLAINTEXT_HOST://localhost:9093'
+      KAFKA_LISTENERS: 'SASL_PLAINTEXT://kafka-sasl:29093,SASL_PLAINTEXT_HOST://0.0.0.0:9093,CONTROLLER://kafka-sasl:29094'
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_PLAINTEXT_HOST:SASL_PLAINTEXT'
+      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
+      KAFKA_INTER_BROKER_LISTENER_NAME: 'SASL_PLAINTEXT'
+      KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: 'PLAIN'
+      KAFKA_SASL_ENABLED_MECHANISMS: 'PLAIN'
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_SUPER_USERS: 'User:admin'
+      KAFKA_OPTS: >-
+        -Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf
+    volumes:
+      - ./resources/kafka_server_jaas.conf:/etc/kafka/kafka_server_jaas.conf:ro
+      - ./resources/sasl_client.properties:/etc/kafka/sasl_client.properties:ro
+    healthcheck:
+      test: >-
+        kafka-broker-api-versions
+        --bootstrap-server localhost:9093
+        --command-config /etc/kafka/sasl_client.properties
+        || exit 1
+      interval: 10s
+      timeout: 10s
+      retries: 12
+    networks:
+      - e2e-sasl-network
+
+networks:
+  e2e-sasl-network:
+    driver: bridge

e2e/resources/jikkou-e2e-sasl.conf

@@ -0,0 +1,21 @@
+jikkou {
+  # Apache Kafka Provider (SASL_PLAINTEXT)
+  provider.kafka {
+    enabled = true
+    type = io.streamthoughts.jikkou.kafka.KafkaExtensionProvider
+    default = true
+    config = {
+      client {
+        bootstrap.servers = "localhost:9093"
+        security.protocol = "SASL_PLAINTEXT"
+        sasl.mechanism = "PLAIN"
+        sasl.jaas.config = "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin-secret\";"
+      }
+      brokers {
+        waitForEnabled = true
+        waitForMinAvailable = 1
+        waitForTimeoutMs = 60000
+      }
+    }
+  }
+}

e2e/resources/kafka-topics-sasl.yaml

@@ -0,0 +1,13 @@
+---
+apiVersion: "kafka.jikkou.io/v1"
+kind: KafkaTopic
+metadata:
+  name: 'e2e-sasl-topic-1'
+  labels:
+    environment: e2e-sasl
+spec:
+  partitions: 3
+  replicas: 1
+  configs:
+    min.insync.replicas: 1
+    cleanup.policy: 'delete'

e2e/resources/kafka_server_jaas.conf

@@ -0,0 +1,6 @@
+KafkaServer {
+    org.apache.kafka.common.security.plain.PlainLoginModule required
+    username="admin"
+    password="admin-secret"
+    user_admin="admin-secret";
+};

e2e/resources/sasl_client.properties

@@ -0,0 +1,3 @@
+security.protocol=SASL_PLAINTEXT
+sasl.mechanism=PLAIN
+sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-secret";

e2e/run-tests.sh

@@ -19,11 +19,14 @@ set -euo pipefail
 SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 PROJECT_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
 COMPOSE_FILE="${SCRIPT_DIR}/docker-compose-e2e.yml"
+COMPOSE_SASL_FILE="${SCRIPT_DIR}/docker-compose-e2e-sasl.yml"
 E2E_RESOURCES="${SCRIPT_DIR}/resources"
 JIKKOU_HOCON_CONFIG="${E2E_RESOURCES}/jikkou-e2e.conf"
+JIKKOU_SASL_HOCON_CONFIG="${E2E_RESOURCES}/jikkou-e2e-sasl.conf"
 # The CLI uses a JSON context file (like kubeconfig), not the HOCON directly.
 # We generate this at runtime so it contains the correct absolute path.
-JIKKOU_CONFIG=""  # set in setup_config()
+JIKKOU_CONFIG=""       # set in setup_config()
+JIKKOU_SASL_CONFIG=""  # set in setup_config()
 
 # ── Flags ────────────────────────────────────────────────────────────────────
 SKIP_BUILD=false
@@ -78,6 +81,18 @@ setup_config() {
 }
 EOF
   log_ok "Generated CLI context config: ${JIKKOU_CONFIG}"
+
+  JIKKOU_SASL_CONFIG=$(mktemp /tmp/jikkou-e2e-sasl-context-XXXXXX.json)
+  cat > "${JIKKOU_SASL_CONFIG}" <<EOF
+{
+  "currentContext": "e2e-sasl",
+  "e2e-sasl": {
+    "configFile": "${JIKKOU_SASL_HOCON_CONFIG}",
+    "configProps": {}
+  }
+}
+EOF
+  log_ok "Generated SASL CLI context config: ${JIKKOU_SASL_CONFIG}"
 }
 
 # Run jikkou binary with the e2e config.
@@ -94,6 +109,14 @@ run_jikkou_capture() {
   set -e
 }
 
+# Run jikkou binary with the SASL config.
+run_jikkou_sasl_capture() {
+  set +e
+  JIKKOU_OUTPUT=$(JIKKOUCONFIG="${JIKKOU_SASL_CONFIG}" "${JIKKOU_BIN}" "$@" 2>&1)
+  JIKKOU_EXIT=$?
+  set -e
+}
+
 assert_exit_code() {
   local expected=$1
   if [[ "${JIKKOU_EXIT}" -ne "${expected}" ]]; then
@@ -170,6 +193,7 @@ build_native() {
 start_services() {
   log_info "Starting Docker services..."
   docker compose -f "${COMPOSE_FILE}" up -d
+  docker compose -f "${COMPOSE_SASL_FILE}" up -d
 
   log_info "Waiting for services to become healthy..."
   local max_wait=180
@@ -178,31 +202,36 @@ start_services() {
 
   while [[ ${elapsed} -lt ${max_wait} ]]; do
     local kafka_ok=false
+    local kafka_sasl_ok=false
     local sr_ok=false
     local connect_ok=false
 
     if docker inspect --format='{{.State.Health.Status}}' e2e-kafka 2>/dev/null | grep -q healthy; then
       kafka_ok=true
     fi
+    if docker inspect --format='{{.State.Health.Status}}' e2e-kafka-sasl 2>/dev/null | grep -q healthy; then
+      kafka_sasl_ok=true
+    fi
     if docker inspect --format='{{.State.Health.Status}}' e2e-schema-registry 2>/dev/null | grep -q healthy; then
       sr_ok=true
     fi
     if docker inspect --format='{{.State.Health.Status}}' e2e-connect 2>/dev/null | grep -q healthy; then
       connect_ok=true
     fi
 
-    if ${kafka_ok} && ${sr_ok} && ${connect_ok}; then
+    if ${kafka_ok} && ${kafka_sasl_ok} && ${sr_ok} && ${connect_ok}; then
       log_ok "All services healthy"
       return 0
     fi
 
-    log_info "Waiting... (${elapsed}s/${max_wait}s) kafka=${kafka_ok} schema-registry=${sr_ok} connect=${connect_ok}"
+    log_info "Waiting... (${elapsed}s/${max_wait}s) kafka=${kafka_ok} kafka-sasl=${kafka_sasl_ok} schema-registry=${sr_ok} connect=${connect_ok}"
     sleep ${interval}
     elapsed=$((elapsed + interval))
   done
 
   log_error "Services did not become healthy within ${max_wait}s"
   docker compose -f "${COMPOSE_FILE}" logs
+  docker compose -f "${COMPOSE_SASL_FILE}" logs
   exit 1
 }
 
@@ -212,6 +241,7 @@ stop_services() {
   else
     log_info "Tearing down Docker services..."
     docker compose -f "${COMPOSE_FILE}" down -v
+    docker compose -f "${COMPOSE_SASL_FILE}" down -v
   fi
 }
 
@@ -583,6 +613,64 @@ EOF
   assert_output_not_contains "e2e-file-sink" || return 1
 }
 
+# ── Kafka Topics over SASL (create, read, delete) ───────────────────────────
+
+test_sasl_kafka_topics_create() {
+  run_jikkou_sasl_capture apply --files "${E2E_RESOURCES}/kafka-topics-sasl.yaml"
+  assert_exit_code 0 || return 1
+
+  run_jikkou_sasl_capture get kafkatopics --name 'e2e-sasl-topic-1'
+  assert_exit_code 0 || return 1
+  assert_output_contains "e2e-sasl-topic-1" || return 1
+}
+
+test_sasl_kafka_topics_read() {
+  run_jikkou_sasl_capture get kafkatopics
+  assert_exit_code 0 || return 1
+  assert_output_contains "e2e-sasl-topic-1" || return 1
+
+  run_jikkou_sasl_capture get kafkatopics --name 'e2e-sasl-topic-1'
+  assert_exit_code 0 || return 1
+  assert_output_contains "partitions.*3" || return 1
+  assert_output_contains "cleanup.policy" || return 1
+}
+
+test_sasl_kafka_topics_delete() {
+  local tmpfile
+  tmpfile=$(mktemp /tmp/e2e-sasl-delete-XXXXXX.yaml)
+  trap "rm -f ${tmpfile}" RETURN
+
+  cat > "${tmpfile}" <<'EOF'
+---
+apiVersion: "kafka.jikkou.io/v1"
+kind: KafkaTopic
+metadata:
+  name: 'e2e-sasl-topic-1'
+  labels:
+    environment: e2e-sasl
+  annotations:
+    jikkou.io/delete: true
+spec:
+  partitions: 3
+  replicas: 1
+  configs:
+    min.insync.replicas: 1
+    cleanup.policy: 'delete'
+EOF
+
+  run_jikkou_sasl_capture apply --files "${tmpfile}"
+  assert_exit_code 0 || return 1
+
+  run_jikkou_sasl_capture get kafkatopics --name 'e2e-sasl-topic-1'
+  assert_output_not_contains "e2e-sasl-topic-1" || return 1
+}
+
+test_sasl_health() {
+  run_jikkou_sasl_capture health get kafka
+  assert_exit_code 0 || return 1
+  assert_output_contains "UP|HEALTHY" || return 1
+}
+
 # ── Health ───────────────────────────────────────────────────────────────────
 
 test_health_indicators() {
@@ -646,6 +734,12 @@ main() {
   run_test "Kafka Connect: update"          test_kafka_connect_update
   run_test "Kafka Connect: delete"          test_kafka_connect_delete
 
+  # ── Kafka Topics over SASL (issue #726) ──
+  run_test "SASL Kafka Topics: create"     test_sasl_kafka_topics_create
+  run_test "SASL Kafka Topics: read"       test_sasl_kafka_topics_read
+  run_test "SASL Kafka Topics: delete"     test_sasl_kafka_topics_delete
+  run_test "SASL Health indicators"        test_sasl_health
+
   # ── Health ──
   run_test "Health indicators"              test_health_indicators
 
@@ -669,8 +763,9 @@ main() {
 
   stop_services
 
-  # Clean up temp config file
+  # Clean up temp config files
   rm -f "${JIKKOU_CONFIG}"
+  rm -f "${JIKKOU_SASL_CONFIG}"
 
   if [[ ${TESTS_FAILED} -gt 0 ]]; then
     exit 1

providers/jikkou-provider-kafka/src/main/resources/META-INF/native-image/io.streamthoughts/jikkou-extension-kafka/reachability-metadata.json

@@ -389,6 +389,20 @@
       "type": "sun.security.krb5.KrbException",
       "allDeclaredMethods": true,
       "allDeclaredConstructors": true
+    },
+    {
+      "type": "javax.security.auth.Subject",
+      "allDeclaredMethods": true
+    },
+    {
+      "type": "org.apache.kafka.common.security.plain.internals.PlainSaslClient$PlainSaslClientFactory",
+      "allDeclaredMethods": true,
+      "allDeclaredConstructors": true
+    },
+    {
+      "type": "org.apache.kafka.common.security.plain.PlainSaslClientProvider",
+      "allDeclaredMethods": true,
+      "allDeclaredConstructors": true
     }
   ],
   "resources": [