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
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"
}
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 3 \
    --timeout-ms 3000 | jq

returns 2 events

{
  "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.

step-08-red-cars-interceptor.json:

{
  "kind" : "Interceptor",
  "apiVersion" : "gateway/v2",
  "metadata" : {
    "name" : "red-cars"
  },
  "spec" : {
    "comment" : "Adding interceptor: red-cars",
    "pluginClass" : "io.conduktor.gateway.interceptor.VirtualSqlTopicPlugin",
    "priority" : 100,
    "config" : {
      "virtualTopic" : "red-cars",
      "statement" : "SELECT type, price as money FROM cars WHERE color = 'red'"
    }
  }
}
curl \
    --silent \
    --request PUT "http://localhost:8888/gateway/v2/interceptor" \
    --header "Content-Type: application/json" \
    --user "admin:conduktor" \
    --data @step-08-red-cars-interceptor.json | jq

Listing interceptors​

Listing interceptors on gateway1

curl \
    --silent \
    --request GET "http://localhost:8888/gateway/v2/interceptor" \
    --user "admin:conduktor" | 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 \
    --max-messages 2 \
    --timeout-ms 3000 | jq

returns 1 event

{
  "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!