Skip to main content
Quick navigation

What is sql topics?

Don't reinvent the wheel to filter and project your messages, just use SQL!

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 topic cars on gateway1

Creating on gateway1:

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

Producing 2 messages in cars

Produce 2 records to the cars topic: our mock car data for cars.

  • A blue car
  • A red car

Sending 2 events

{
"type" : "Sports",
"price" : 75,
"color" : "blue"
}
{
"type" : "SUV",
"price" : 55,
"color" : "red"
}

with

echo '{"type":"Sports","price":75,"color":"blue"}' | \
kafka-console-producer \
--bootstrap-server localhost:6969 \
--topic cars

echo '{"type":"SUV","price":55,"color":"red"}' | \
kafka-console-producer \
--bootstrap-server localhost:6969 \
--topic cars

Consuming from cars

Let's confirm the 2 records are there by consuming from the cars topic.

kafka-console-consumer \
--bootstrap-server localhost:6969 \
--topic cars \
--from-beginning \
--max-messages 2 \
--timeout-ms 10000 | jq

returns

Processed a total of 2 messages
{
"type": "Sports",
"price": 75,
"color": "blue"
}
{
"type": "SUV",
"price": 55,
"color": "red"
}

Adding interceptor red-cars

Let's create the interceptor to filter out the red cars from the cars topic.

Creating the interceptor named red-cars of the plugin io.conduktor.gateway.interceptor.VirtualSqlTopicPlugin using the following payload

{
"pluginClass" : "io.conduktor.gateway.interceptor.VirtualSqlTopicPlugin",
"priority" : 100,
"config" : {
"virtualTopic" : "red-cars",
"statement" : "SELECT type, price as money FROM cars WHERE color = 'red'"
}
}

Here's how to send it:

curl \
--request POST "http://localhost:8888/admin/interceptors/v1/vcluster/passthrough/interceptor/red-cars" \
--header 'Content-Type: application/json' \
--user 'admin:conduktor' \
--silent \
--data @step-08-red-cars.json | jq

Listing interceptors for passthrough

Listing interceptors on gateway1 for virtual cluster passthrough

curl \
--request GET 'http://localhost:8888/admin/interceptors/v1/vcluster/passthrough' \
--header 'Content-Type: application/json' \
--user 'admin:conduktor' \
--silent | jq

Consume from the virtual topic red-cars

Let's consume from our virtual topic red-cars.

You now see only one car, the red one, please note that its format changed according to our SQL statement's projection.

If you are wondering if you can be a bit more fancy, the answer is ... yes!

kafka-console-consumer \
--bootstrap-server localhost:6969 \
--topic red-cars \
--from-beginning \
--timeout-ms 10000 | jq

returns

Processed a total of 1 messages
{
"type": "SUV",
"money": 55
}

Can we do more with SQL?

Yes! We sure can.

SELECT
type,
price as newPriceName,
color,
CASE
WHEN color = 'red' AND price > 1000 THEN 'Exceptional'
WHEN price > 8000 THEN 'Luxury'
ELSE 'Regular'
END as quality,
record.offset as record_offset,
record.partition as record_partition
FROM cars

is an example where you mix projection, case, renaming, and kafka metadata.

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

SQL topic is really a game changer!