Skip to main content
Quick navigation

Large message handling in Kafka with built-in claimcheck pattern.

View the full demo in realtime​

You can either follow all the steps manually, or watch the recording

Review the docker compose environment​

As can be seen from docker-compose.yaml the demo environment consists of the following services:

  • cli-aws
  • gateway1
  • gateway2
  • kafka-client
  • kafka1
  • kafka2
  • kafka3
  • minio
  • schema-registry
cat docker-compose.yaml

Starting the docker environment​

Start all your docker processes, wait for them to be up and ready, then run in background

  • --wait: Wait for services to be running|healthy. Implies detached mode.
  • --detach: Detached mode: Run containers in the background
docker compose up --detach --wait

Review credentials​

cat credentials

Let's create a bucket​

docker compose exec cli-aws \
  aws \
    --profile minio \
    --endpoint-url=http://minio:9000 \
    --region eu-south-1 \
    s3api create-bucket \
      --bucket bucket

Creating topic large-messages on gateway1​

Creating on gateway1:

  • Topic large-messages with partitions:1 and replication-factor:1
kafka-topics \
    --bootstrap-server localhost:6969 \
    --replication-factor 1 \
    --partitions 1 \
    --create --if-not-exists \
    --topic large-messages

Adding interceptor large-messages​

Let's ask Gateway to offload large messages to S3

step-08-large-messages-interceptor.json:

{
  "kind" : "Interceptor",
  "apiVersion" : "gateway/v2",
  "metadata" : {
    "name" : "large-messages"
  },
  "spec" : {
    "comment" : "Adding interceptor: large-messages",
    "pluginClass" : "io.conduktor.gateway.interceptor.LargeMessageHandlingPlugin",
    "priority" : 100,
    "config" : {
      "topic" : "large-messages",
      "s3Config" : {
        "accessKey" : "minio",
        "secretKey" : "minio123",
        "bucketName" : "bucket",
        "region" : "eu-south-1",
        "uri" : "http://minio:9000"
      }
    }
  }
}
curl \
    --silent \
    --request PUT "http://localhost:8888/gateway/v2/interceptor" \
    --header "Content-Type: application/json" \
    --user "admin:conduktor" \
    --data @step-08-large-messages-interceptor.json | jq

Let's create a large message​

openssl rand -hex $((20*1024*1024)) > large-message.bin 
ls -lh large-message.bin

Sending large pdf file through kafka​

requiredMemory=$(( 2 * $(cat large-message.bin | wc -c | awk '{print $1}')))

kafka-producer-perf-test \
  --topic large-messages \
  --throughput -1 \
  --num-records 1 \
  --payload-file large-message.bin \
  --producer-props \
    bootstrap.servers=localhost:6969 \
    max.request.size=$requiredMemory \
    buffer.memory=$requiredMemory

Let's read the message back​

kafka-console-consumer  \
  --bootstrap-server localhost:6969 \
  --topic large-messages \
  --from-beginning \
  --max-messages 1 > from-kafka.bin

Let's compare the files​

ls -lH *bin

Let's look at what's inside minio​

docker compose exec cli-aws \
    aws \
        --profile minio \
        --endpoint-url=http://minio:9000 \
        --region eu-south-1 \
        s3 \
        ls s3://bucket --recursive --human-readable

Consuming from large-messages​

Consuming from large-messages in cluster kafka1

kafka-console-consumer \
    --bootstrap-server localhost:9092,localhost:9093,localhost:9094 \
    --topic large-messages \
    --from-beginning \
    --max-messages 5 \
    --timeout-ms 3000 \
    --property print.headers=true | jq

Tearing down the docker environment​

Remove all your docker processes and associated volumes

  • --volumes: Remove named volumes declared in the "volumes" section of the Compose file and anonymous volumes attached to containers.
docker compose down --volumes

Conclusion

Large messages can be handled in Kafka using Conduktor!