example/kafka-mesh: Add docs/test for batching and high volume handling (#1034)

Signed-off-by: Ryan Northey <[email protected]>

Author: phlax

kafka-mesh/example.rst

@@ -146,6 +146,63 @@ produce and fetch request counts:
    kafka.kafka_mesh.request.metadata_request: 8
 
 
+Step 8: Test high-volume producing with batched records
+*******************************************************
+
+In production, Kafka producers often batch multiple records into a single
+``ProduceRequest`` to improve throughput. The mesh filter must correctly handle
+these batched requests and route records to the appropriate upstream cluster.
+
+Send multiple messages rapidly to the ``apricots`` topic (which routes to
+``cluster1`` based on the ``a`` prefix). The producer will automatically batch
+these into fewer requests:
+
+.. code-block:: console
+
+   $ docker compose run --rm kafka-client /bin/bash -c " \
+       for i in {1..20}; do \
+           echo \"apricot message \$i\"; \
+       done | kafka-console-producer --request-required-acks 1 --producer-property enable.idempotence=false --broker-list proxy:10000 --topic apricots"
+
+Now verify that all 20 messages arrived at ``cluster1`` by consuming directly
+from the upstream:
+
+.. code-block:: console
+
+   $ docker compose run --rm kafka-client \
+       kafka-console-consumer --bootstrap-server kafka-cluster1:9092 --topic apricots --from-beginning --max-messages 20 | wc -l
+   20
+
+This confirms that even though the producer may have batched the records into
+multiple ``ProduceRequest``, the mesh filter correctly routed all messages to the
+appropriate cluster. This is critical for high-throughput production workloads.
+
+Next, send 20 messages rapidly to the ``blueberries`` topic (which routes to
+cluster2 based on the ``b`` prefix). This demonstrates that the mesh filter
+correctly splits batched ProduceRequests when they need to go to different
+upstream clusters:
+
+.. code-block:: console
+
+   $ docker compose run --rm kafka-client /bin/bash -c " \
+       for i in {1..20}; do \
+           echo \"blueberry message \$i\"; \
+       done | kafka-console-producer --request-required-acks 1 --producer-property enable.idempotence=false --broker-list proxy:10000 --topic blueberries"
+
+Verify that all 20 messages arrived at cluster2:
+
+.. code-block:: console
+
+   $ docker compose run --rm kafka-client \
+       kafka-console-consumer --bootstrap-server kafka-cluster2:9092 --topic blueberries --from-beginning --max-messages 20 | wc -l
+   20
+
+This demonstrates the mesh filter's ability to handle batched records destined
+for different clusters. In production environments where a single producer
+sends to multiple topics across different clusters, this ensures records are
+correctly routed based on their destination.
+
+
 .. seealso::
 
    :ref:`Envoy Kafka mesh filter <config_network_filters_kafka_mesh>`

kafka-mesh/verify.sh

@@ -36,3 +36,57 @@ stats_output=$(_curl "http://localhost:${PORT_ADMIN}/stats?filter=kafka")
 echo "$stats_output" | grep "produce_request" | grep -v ": 0"
 echo "$stats_output" | grep "fetch_request" | grep -v ": 0"
 echo "$stats_output" | grep "metadata_request" | grep -v ": 0"
+
+run_log "Test high-volume producing with batched records"
+# Send 20 messages rapidly to trigger producer batching
+kafka_client /bin/bash -c " \
+    for i in {1..20}; do \
+        echo \"apricot message \$i\"; \
+    done | kafka-console-producer --request-required-acks 1 --producer-property enable.idempotence=false --broker-list proxy:10000 --topic apricots"
+
+run_log "Verify all 20 messages arrived at cluster1"
+# Consume all messages and count them
+message_count=$(kafka_client kafka-console-consumer --bootstrap-server kafka-cluster1:9092 --topic apricots --from-beginning --max-messages 20 2>/dev/null | wc -l)
+message_count=${message_count:-0}
+run_log "Received $message_count messages from apricots topic"
+
+if [[ "$message_count" -eq 20 ]]; then
+    run_log "SUCCESS: All 20 messages arrived at cluster1"
+else
+    echo "ERROR: Expected 20 messages but received $message_count" >&2
+    exit 1
+fi
+
+run_log "Verify produce metrics reflect the batched requests"
+# Get the produce_request count - it should be greater than 0 and likely less than 20 (due to batching)
+stats_output=$(_curl "http://localhost:${PORT_ADMIN}/stats?filter=kafka.kafka_mesh.request.produce_request")
+produce_count=$(echo "$stats_output" | grep "produce_request:" | cut -f2 -d':' | tr -d ' ')
+produce_count=${produce_count:-0}
+run_log "Total produce_request count: $produce_count"
+
+if [[ "$produce_count" -gt 0 ]]; then
+    run_log "SUCCESS: Produce requests tracked correctly (count: $produce_count)"
+else
+    echo "ERROR: No produce requests tracked" >&2
+    exit 1
+fi
+
+run_log "Test high-volume producing to second cluster with batched records"
+# Send 20 messages rapidly to blueberries topic (routes to cluster2)
+kafka_client /bin/bash -c " \
+    for i in {1..20}; do \
+        echo \"blueberry message \$i\"; \
+    done | kafka-console-producer --request-required-acks 1 --producer-property enable.idempotence=false --broker-list proxy:10000 --topic blueberries"
+
+run_log "Verify all 20 messages arrived at cluster2"
+# Consume all messages and count them
+message_count=$(kafka_client kafka-console-consumer --bootstrap-server kafka-cluster2:9092 --topic blueberries --from-beginning --max-messages 20 2>/dev/null | wc -l)
+message_count=${message_count:-0}
+run_log "Received $message_count messages from blueberries topic"
+
+if [[ "$message_count" -eq 20 ]]; then
+    run_log "SUCCESS: All 20 messages arrived at cluster2"
+else
+    echo "ERROR: Expected 20 messages but received $message_count" >&2
+    exit 1
+fi