Add kafka-mesh example demonstrating topic-based routing with mesh filter

[Copilot]

Plan for kafka-mesh Example

• Create kafka-mesh directory structure
• Create Dockerfile-kafka (copy from kafka example)
• Create Dockerfile-zookeeper (copy from kafka example)
• Create envoy.yaml with mesh filter configuration
• Create docker-compose.yaml with two Kafka clusters
• Create example.rst documentation
• Create verify.sh test script
• Create README.md
• Update BUILD file to add kafka-mesh to EXAMPLE_TESTS
• Update BUILD file to add kafka-mesh to contrib_configs
• Update .github/dependabot.yml to add kafka-mesh docker dependencies
• Run code review and address feedback
• Run security scan
• Fix producer idempotence issue for mesh filter compatibility
• Fix consumer group coordination issue for mesh filter compatibility
• Use direct partition assignment to avoid coordinator APIs

Summary

Successfully added the kafka-mesh example demonstrating the Kafka mesh filter's routing capabilities. This example is designed to replace existing Envoy Kafka integration tests.

Changes

• Added --request-required-acks 1 --producer-property enable.idempotence=false to producer commands to disable transactional features
• Changed consumer commands to use --partition 0 instead of consumer groups to avoid FIND_COORDINATOR API (not supported by mesh filter)
• Added documentation note explaining mesh filter only supports Produce and Fetch APIs

Security Summary

No security vulnerabilities detected.

Original prompt

Summary

Add a new kafka-mesh example that demonstrates the Kafka mesh filter's ability to route Kafka traffic to different upstream clusters based on topic prefix.

This is part of migrating Kafka integration test coverage from envoyproxy/envoy to the examples repo. See: envoyproxy/envoy#43182

What the mesh filter does

Unlike the existing kafka example (broker filter) which passes all traffic to a single upstream with metrics, the mesh filter:

• Inspects the Kafka protocol
• Extracts topic names from produce/fetch requests
• Routes to different upstream Kafka clusters based on configurable rules

Example design

The example should demonstrate routing with two upstream Kafka clusters:

• Topics starting with a (e.g., apples) → cluster1
• Topics starting with b (e.g., bananas) → cluster2

Files to create

kafka-mesh/example.rst

.. _install_sandboxes_kafka_mesh:

Kafka mesh filter
=================

.. sidebar:: Requirements

   .. include:: _include/docker-env-setup-link.rst

   :ref:`curl <start_sandboxes_setup_curl>`
        Used to make HTTP requests.

This example demonstrates the Kafka mesh filter, which routes Kafka traffic 
to different upstream clusters based on topic name.

The :ref:`Kafka broker filter <install_sandboxes_kafka>` passes all traffic 
to a single upstream Kafka cluster while collecting metrics. The mesh filter 
goes further: it reads the Kafka protocol, extracts the topic name from each 
request, and forwards it to the appropriate upstream cluster.

In this example we configure two upstream Kafka clusters with routing rules:

- Topics starting with ``a`` (e.g., ``apples``) → **cluster1**
- Topics starting with ``b`` (e.g., ``bananas``) → **cluster2**

Clients connect to Envoy as if it were a single Kafka broker. Envoy handles 
the routing transparently.


Step 1: Start all containers
****************************

Change to the ``kafka-mesh`` directory and start the containers. This brings 
up Envoy, two independent Kafka clusters, and a shared Zookeeper instance.

.. code-block:: console

   $ pwd
   envoy/examples/kafka-mesh
   $ docker compose pull
   $ docker compose up --build -d
   $ docker compose ps

         Name                       State              Ports
   ------------------------------------------------------------------
   kafka-mesh_proxy_1          Up             0.0.0.0:10000->10000/tcp
   kafka-mesh_kafka-cluster1_1 Up             9092/tcp
   kafka-mesh_kafka-cluster2_1 Up             9092/tcp
   kafka-mesh_zookeeper_1      Up (healthy)   2181/tcp


Step 2: Produce a message to the ``apples`` topic
*************************************************

Send a message to a topic starting with ``a``. The mesh filter will route 
this to **cluster1**.

.. code-block:: console

   $ docker compose run --rm kafka-client \
       /bin/bash -c "echo 'hello from apples' | kafka-console-producer --broker-list proxy:10000 --topic apples"


Step 3: Produce a message to the ``bananas`` topic
**************************************************

Send a message to a topic starting with ``b``. The mesh filter will route 
this to **cluster2**.

.. code-block:: console

   $ docker compose run --rm kafka-client \
       /bin/bash -c "echo 'hello from bananas' | kafka-console-producer --broker-list proxy:10000 --topic bananas"


Step 4: Verify the message landed in cluster1
*********************************************

To confirm the routing worked, consume directly from **cluster1** (bypassing 
Envoy). You should see the ``apples`` message:

.. code-block:: console

   $ docker compose run --rm kafka-client \
       kafka-console-consumer --bootstrap-server kafka-cluster1:9092 --topic apples --from-beginning --max-messages 1
   hello from apples

The ``bananas`` topic should not exist on cluster1.


Step 5: Verify the message landed in cluster2
*********************************************

Similarly, consume directly from **cluster2**. You should see the ``bananas`` 
message:

.. code-block:: console

   $ docker compose run --rm kafka-client \
       kafka-console-consumer --bootstrap-server kafka-cluster2:9092 --topic bananas --from-beginning --max-messages 1
   hello from bananas

The ``apples`` topic should not exist on cluster2.


Step 6: Consume through the mesh filter
***************************************

The mesh filter also routes fetch (consume) requests to the correct upstream. 
Clients can consume through Envoy without knowing which cluster holds the data.

Consume the ``apples`` topic through Envoy:

.. code-block:: console

   $ docker compose run --rm kafka-client \
       kafka-console-consumer --bootstrap-server proxy:10000 --topic apples --from-beginning --max-messages 1
   hello from apples

Consume the ``bananas`` topic through Envoy:

.. code-block:: console

   $ docker compose run --rm kafka-client \
       kafka-console-consumer --bootstrap-server proxy:10000 --topic bananas --from...

</details>



<!-- START COPILOT CODING AGENT SUFFIX -->

*This pull request was created from Copilot chat.*
>

<!-- START COPILOT CODING AGENT TIPS -->
---

✨ Let Copilot coding agent [set things up for you](https://github.com/envoyproxy/examples/issues/new?title=✨+Set+up+Copilot+instructions&body=Configure%20instructions%20for%20this%20repository%20as%20documented%20in%20%5BBest%20practices%20for%20Copilot%20coding%20agent%20in%20your%20repository%5D%28https://gh.io/copilot-coding-agent-tips%29%2E%0A%0A%3COnboard%20this%20repo%3E&assignees=copilot) — coding agent works faster and does higher quality work when set up for your repo.

[phlax]

@copilot fails with

[kafka-mesh] Produce message to topic 'apples' (routes to cluster1)
^@[2026-01-27 11:11:20,027] ERROR [Producer clientId=console-producer] Aborting producer batches due to fatal error (org.apache.kafka.clients.producer.internals.Sender)
org.apache.kafka.common.errors.UnsupportedVersionException: The node does not support INIT_PRODUCER_ID
[2026-01-27 11:11:20,031] ERROR Error when sending message to topic apples with key: null, value: 17 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
org.apache.kafka.common.errors.UnsupportedVersionException: The node does not support INIT_PRODUCER_ID
org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state
at org.apache.kafka.clients.producer.internals.TransactionManager.maybeFailWithError(TransactionManager.java:1041)
at org.apache.kafka.clients.producer.internals.TransactionManager.maybeAddPartition(TransactionManager.java:385)
at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:1090)
at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:991)
at kafka.tools.ConsoleProducer$.send(ConsoleProducer.scala:118)
at kafka.tools.ConsoleProducer$.loopReader(ConsoleProducer.scala:91)
at kafka.tools.ConsoleProducer$.main(ConsoleProducer.scala:100)
at kafka.tools.ConsoleProducer.main(ConsoleProducer.scala)
Caused by: org.apache.kafka.common.errors.UnsupportedVersionException: The node does not support INIT_PRODUCER_ID

FAILURE DEBUG

[Copilot]

@copilot fails with

[kafka-mesh] Produce message to topic 'apples' (routes to cluster1)...

Fixed in b29a8d3. The mesh filter doesn't support INIT_PRODUCER_ID (transactional/idempotent producer), so added --request-required-acks 1 --producer-property enable.idempotence=false flags to disable idempotence in both the verify.sh script and documentation.

[phlax]

@copilot next

> [kafka-mesh] Consume 'apples' through mesh filter
[2026-01-27 11:32:53,174] ERROR Error processing message, terminating consumer process:  (org.apache.kafka.tools.consumer.ConsoleConsumer)
org.apache.kafka.common.errors.UnsupportedVersionException: The node does not support FIND_COORDINATOR
Processed a total of 0 messages
FAILURE DEBUG
DISK SPACE
Filesystem      Size  Used Avail Use% Mounted on
overlay         145G   61G   84G  42% /
tmpfs            64M     0   64M   0% /dev
shm              64M     0   64M   0% /dev/shm
/dev/root       145G   61G   84G  42% /cleanup.sh

[Copilot]

@copilot next

Fixed in bad4e7a. The mesh filter only supports Produce and Fetch APIs, not consumer group coordination (FIND_COORDINATOR). Added --consumer-property group.id= to disable consumer groups when consuming through the mesh filter. Direct consumption from cluster1/cluster2 still works normally.

[phlax]

@copilot still no apples

> [kafka-mesh] Consume 'apples' through mesh filter
[2026-01-27 11:55:11,910] WARN [Consumer clientId=console-consumer, groupId=] Support for using the empty group id by consumers is deprecated and will be removed in the next major release. (org.apache.kafka.clients.consumer.internals.LegacyKafkaConsumer)
[2026-01-27 11:55:12,892] ERROR Error processing message, terminating consumer process:  (org.apache.kafka.tools.consumer.ConsoleConsumer)
org.apache.kafka.common.errors.UnsupportedVersionException: The node does not support FIND_COORDINATOR
Processed a total of 0 messages
FAILURE DEBUG
DISK SPACE
Filesystem      Size  Used Avail Use% Mounted on
overlay          74G  5.7G   64G   9% /
tmpfs            64M     0   64M   0% /dev
shm              64M     0   64M   0% /dev/shm
/dev/root        72G   63G  9.3G  88% /cleanup.sh
/dev/sdb1        74G  5.7G   64G   9% /etc/hosts
DOCKER COMPOSE LOGS

[Copilot]

@copilot still no apples

Fixed in ef83382. Using group.id= (empty string) is deprecated and still tries to use FIND_COORDINATOR. Changed to --partition 0 to directly consume from the partition without any consumer group coordination APIs.