[kafka_log] Update integration to input type to provide multi-signal support

packages/kafka_log/_dev/deploy/docker/docker-compose.yml

@@ -1,21 +1,33 @@
-version: '2.3'
 services:
   kafka-service:
-    image: bashj79/kafka-kraft
-    healthcheck:
-      test: nc -z kafka-service 9094 || exit -1
-      interval: 10s
-      timeout: 5s
-      retries: 3
-      start_period: 10s
+    # Official Apache Kafka image (KRaft). See https://hub.docker.com/r/apache/kafka
+    image: apache/kafka:4.1.2
     hostname: kafka-service
-    environment:
-      KAFKA_LISTENERS: "INTERNAL://kafka-service:9092,EXTERNAL://:9094, CONTROLLER://:9093"
-      KAFKA_ADVERTISED_LISTENERS: "INTERNAL://kafka-service:9092,EXTERNAL://kafka-service:9094"
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT, CONTROLLER:PLAINTEXT"
-      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
     ports:
       - 9094
+    environment:
+      KAFKA_NODE_ID: 1
+      KAFKA_PROCESS_ROLES: broker,controller
+      # Client traffic on 9094; controller on 9093 (single-node KRaft quorum)
+      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9094,CONTROLLER://0.0.0.0:9093
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-input-svc:9094
+      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-input-svc:9093
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+    healthcheck:
+      test:
+        [
+          "CMD-SHELL",
+          "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server 127.0.0.1:9094 || exit 1",
+        ]
+      interval: 10s
+      timeout: 10s
+      retries: 6
+      start_period: 45s
   kafka-generic:
     image: docker.elastic.co/observability/stream:v0.18.0
     volumes:

packages/kafka_log/_dev/test/system/test-kafka-config.yml

@@ -0,0 +1,9 @@
+service: kafka-service
+input: kafka
+vars:
+  data_stream.dataset: kafka_log.generic
+  topics:
+    - testTopic
+  hosts:
+    - "{{Hostname}}:{{Port}}"
+  group_id: system_test

packages/kafka_log/changelog.yml

@@ -1,3 +1,7 @@
+- version: "2.0.0"
+  changes:
+    - description: Convert kafka_log to an input package. Update the kibana constraint to ^9.4.0.
+      type: enhancement
 - version: "1.9.1"
   changes:
     - description: Updated the version field description to clarify default protocol versions.

packages/kafka_log/docs/README.md

@@ -1,16 +1,14 @@
-# Custom Kafka Log integration
+# Custom Kafka Log input package
 
-The custom Kafka log integration is used to read from topics in a Kafka cluster.
-
-To configure this integration, specify a list of one or more hosts in the cluster to bootstrap the connection with, a list of topics to track, and a group_id for the connection.
+This package is an **input** type integration: it exposes the Filebeat Kafka input in Fleet so you can read from topics in a Kafka cluster and choose a dataset name for routing.
 
+Configure bootstrap **hosts**, **topics**, and a consumer **group_id** (and optional SASL, TLS, parsers, and related settings) in the policy.
 
 ## Compatibility
-This Integration works with all Kafka versions in between 0.11 and 2.8.0. Older versions might work as well, but are not supported.
-
+This package works with Kafka versions between 0.11 and 2.8.0. Earlier versions might work as well, but are not supported.
 
 ## Ingest Pipelines
-Custom ingest pipelines may be added by adding the name to the pipeline configuration option, creating custom ingest pipelines can be done either through the API or the [Ingest Node Pipeline UI](/app/management/ingest/ingest_pipelines/).
+Custom ingest pipelines can be added by setting the pipeline option, you can create pipelines using the API or the [Ingest Node Pipeline UI](/app/management/ingest/ingest_pipelines/).
 
 **ECS Field Reference**
 

packages/kafka_log/data_stream/generic/fields/base-fields.yml → packages/kafka_log/fields/base-fields.yml

@@ -14,7 +14,6 @@
 - name: event.dataset
   type: constant_keyword
   description: Event dataset
-  value: kafka_log.generic
 - name: "@timestamp"
   type: date
   description: Event timestamp.