はじめに
Kong Gateway 3.10 でKafkaへの接続機能ができました(Enterprise限定)
今回はkafka-upstreamプラグインを使ってみます。
セットアップ
Enterprise限定の機能なので Konnect を用いてDPを立てていきます
Networkを作成し、Kongを繋げます
docker network create kong_kafka
docker network connect kong_kafka reverent_lovelace
Kafkaを立てます
docker run -d --name kafka \
-e KAFKA_NODE_ID=1 \
-e KAFKA_PROCESS_ROLES=broker,controller \
-e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT \
-e KAFKA_LISTENERS=PLAINTEXT://:29092,CONTROLLER://:9093,PLAINTEXT_HOST://:9092 \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092 \
-e KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT \
-e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
-e KAFKA_CONTROLLER_QUORUM_VOTERS=1@kafka:9093 \
-e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
-e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 \
-e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \
-e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \
-e KAFKA_NUM_PARTITIONS=3 \
--network kong_kafka \
apache/kafka:4.1.1
Kafka Topicを作ります
docker exec -it kafka sh
/ $ /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:29092 --create --topic test-topic
KonnectでRouteとPluginを作成します
Route
{"data":[{"created_at":1763632119,"https_redirect_status_code":426,"id":"df163c03-c7c1-4408-8a04-134aa2dbd5f6","name":"Kafka","path_handling":"v0","paths":["/"],"preserve_host":false,"protocols":["http","https"],"regex_priority":0,"request_buffering":true,"response_buffering":true,"strip_path":true,"updated_at":1763632119}]}
Plugin
{"data":[{"config":{"allowed_topics":null,"authentication":{"mechanism":null,"password":null,"strategy":null,"tokenauth":null,"user":null},"bootstrap_servers":[{"host":"kafka","port":29092}],"cluster_name":"ACERkaRZu5qwE16O695t7aCryReYTuPf","forward_body":true,"forward_headers":false,"forward_method":false,"forward_uri":false,"keepalive":60000,"keepalive_enabled":false,"key_query_arg":null,"message_by_lua_functions":null,"producer_async":true,"producer_async_buffering_limits_messages_in_memory":50000,"producer_async_flush_timeout":1000,"producer_request_acks":1,"producer_request_limits_bytes_per_request":1048576,"producer_request_limits_messages_per_request":200,"producer_request_retries_backoff_timeout":100,"producer_request_retries_max_attempts":10,"producer_request_timeout":2000,"schema_registry":{"confluent":null},"security":{"certificate_id":null,"ssl":null},"timeout":10000,"topic":"test-topic","topics_query_arg":null},"created_at":1763633224,"enabled":true,"id":"ccf55cbd-9062-4cbc-a346-a6b556c447e1","name":"kafka-upstream","protocols":["grpc","grpcs","http","https"],"route":{"id":"df163c03-c7c1-4408-8a04-134aa2dbd5f6"},"updated_at":1763637196}]}
Kafka Pluginは Topic と Bootstrap Server のみを設定しています
Bootstrap Server については Kafka containerに設定した通り kafka:29092としています
投げてみる
curl -X POST localhost:8000/ -d "test data"
{"message":"message sent"}
届いたっぽい
/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server kafka:29092 --topic test-topic
{"body":"test data","body_args":{"test data":true}}
body , body_args という形式でのJSONになって登録されているみたいです