Skip to main content
Quick navigation

Dynamic Header Injection & Removal

There are multiple interceptors available for manipulating headers, either injection or regex based removal.

This demo will run you through some of these use cases step-by-step.

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:

  • gateway1
  • gateway2
  • kafka-client
  • kafka1
  • kafka2
  • kafka3
  • schema-registry
  • zookeeper
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

Creating virtual cluster teamA

Creating virtual cluster teamA on gateway gateway1 and reviewing the configuration file to access it

# Generate virtual cluster teamA with service account sa
token=$(curl \
--request POST "http://localhost:8888/admin/vclusters/v1/vcluster/teamA/username/sa" \
--header 'Content-Type: application/json' \
--user 'admin:conduktor' \
--silent \
--data-raw '{"lifeTimeSeconds": 7776000}' | jq -r ".token")

# Create access file
echo """
bootstrap.servers=localhost:6969
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='sa' password='$token';
""" > teamA-sa.properties

# Review file
cat teamA-sa.properties

Creating topic users on teamA

Creating on teamA:

  • Topic users with partitions:1 and replication-factor:1
kafka-topics \
--bootstrap-server localhost:6969 \
--command-config teamA-sa.properties \
--replication-factor 1 \
--partitions 1 \
--create --if-not-exists \
--topic users

Adding interceptor inject-headers

Let's create the interceptor to inject various headers

cat step-07-inject-headers.json | jq

curl \
--request POST "http://localhost:8888/admin/interceptors/v1/vcluster/teamA/interceptor/inject-headers" \
--header 'Content-Type: application/json' \
--user 'admin:conduktor' \
--silent \
--data @step-07-inject-headers.json | jq

Send tom and laura into users

Producing 2 messages in users in cluster teamA

Sending 2 events

{
"name" : "tom",
"username" : "tom@conduktor.io",
"password" : "motorhead",
"visa" : "#abc123",
"address" : "Chancery lane, London"
}
{
"name" : "laura",
"username" : "laura@conduktor.io",
"password" : "kitesurf",
"visa" : "#888999XZ",
"address" : "Dubai, UAE"
}

with

echo '{"name":"tom","username":"tom@conduktor.io","password":"motorhead","visa":"#abc123","address":"Chancery lane, London"}' | \
kafka-console-producer \
--bootstrap-server localhost:6969 \
--producer.config teamA-sa.properties \
--topic users

echo '{"name":"laura","username":"laura@conduktor.io","password":"kitesurf","visa":"#888999XZ","address":"Dubai, UAE"}' | \
kafka-console-producer \
--bootstrap-server localhost:6969 \
--producer.config teamA-sa.properties \
--topic users

Verify tom and laura have the corresponding headers

Verify tom and laura have the corresponding headers in cluster teamA

kafka-console-consumer \
--bootstrap-server localhost:6969 \
--consumer.config teamA-sa.properties \
--topic users \
--from-beginning \
--max-messages 2 \
--timeout-ms 10000 \
--property print.headers=true | jq

returns 2 events

{
"headers" : {
"X-INTERPOLATED" : "User sa via ip 192.168.65.1",
"X-MY-KEY" : "my own value",
"X-USER" : "sa"
},
"value" : {
"name" : "tom",
"username" : "tom@conduktor.io",
"password" : "motorhead",
"visa" : "#abc123",
"address" : "Chancery lane, London"
}
}
{
"headers" : {
"X-INTERPOLATED" : "User sa via ip 192.168.65.1",
"X-MY-KEY" : "my own value",
"X-USER" : "sa"
},
"value" : {
"name" : "laura",
"username" : "laura@conduktor.io",
"password" : "kitesurf",
"visa" : "#888999XZ",
"address" : "Dubai, UAE"
}
}

Adding interceptor remove-headers

Let's create the interceptor remove-headers to remove headers that match X-MY-.*

cat step-10-remove-headers.json | jq

curl \
--request POST "http://localhost:8888/admin/interceptors/v1/vcluster/teamA/interceptor/remove-headers" \
--header 'Content-Type: application/json' \
--user 'admin:conduktor' \
--silent \
--data @step-10-remove-headers.json | jq

Verify tom and laura have the corresponding headers

Verify tom and laura have the corresponding headers in cluster teamA

kafka-console-consumer \
--bootstrap-server localhost:6969 \
--consumer.config teamA-sa.properties \
--topic users \
--from-beginning \
--max-messages 2 \
--timeout-ms 10000 \
--property print.headers=true | jq

returns 2 events

{
"headers" : {
"X-INTERPOLATED" : "User sa via ip 192.168.65.1",
"X-USER" : "sa"
},
"value" : {
"name" : "tom",
"username" : "tom@conduktor.io",
"password" : "motorhead",
"visa" : "#abc123",
"address" : "Chancery lane, London"
}
}
{
"headers" : {
"X-INTERPOLATED" : "User sa via ip 192.168.65.1",
"X-USER" : "sa"
},
"value" : {
"name" : "laura",
"username" : "laura@conduktor.io",
"password" : "kitesurf",
"visa" : "#888999XZ",
"address" : "Dubai, UAE"
}
}

Remove interceptor remove-headers

Let's delete the interceptor remove-headers so we can access all our headers again

curl \
--request DELETE "http://localhost:8888/admin/interceptors/v1/vcluster/teamA/interceptor/remove-headers" \
--header 'Content-Type: application/json' \
--user 'admin:conduktor' \
--silent | jq

Verify tom and laura have X-MY-KEY back

Verify tom and laura have X-MY-KEY back in cluster teamA

kafka-console-consumer \
--bootstrap-server localhost:6969 \
--consumer.config teamA-sa.properties \
--topic users \
--from-beginning \
--max-messages 2 \
--timeout-ms 10000 \
--property print.headers=true | jq

returns 2 events

{
"headers" : {
"X-INTERPOLATED" : "User sa via ip 192.168.65.1",
"X-MY-KEY" : "my own value",
"X-USER" : "sa"
},
"value" : {
"name" : "tom",
"username" : "tom@conduktor.io",
"password" : "motorhead",
"visa" : "#abc123",
"address" : "Chancery lane, London"
}
}
{
"headers" : {
"X-INTERPOLATED" : "User sa via ip 192.168.65.1",
"X-MY-KEY" : "my own value",
"X-USER" : "sa"
},
"value" : {
"name" : "laura",
"username" : "laura@conduktor.io",
"password" : "kitesurf",
"visa" : "#888999XZ",
"address" : "Dubai, UAE"
}
}

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

Leveraging headers in Kafka is of tremendous help!