Quick navigation
Yes, SQL topic works with Avro/ProtoBuf
View the full demo in realtime
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
- Command
- File Content
cat docker-compose.yaml
services:
kafka1:
image: confluentinc/cp-server:7.5.0
hostname: kafka1
container_name: kafka1
ports:
- 9092:9092
environment:
KAFKA_NODE_ID: 1
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
KAFKA_LISTENERS: INTERNAL://kafka1:29092,CONTROLLER://kafka1:29093,EXTERNAL://0.0.0.0:9092
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka1:29092,EXTERNAL://localhost:9092
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka1:29093,2@kafka2:29093,3@kafka3:29093
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
KAFKA_LOG4J_ROOT_LOGLEVEL: WARN
KAFKA_AUTO_CREATE_TOPICS_ENABLE: false
CLUSTER_ID: p0KPFA_mQb2ixdPbQXPblw
healthcheck:
test: nc -zv kafka1 29092 || exit 1
interval: 5s
retries: 25
kafka2:
image: confluentinc/cp-server:7.5.0
hostname: kafka2
container_name: kafka2
ports:
- 9093:9093
environment:
KAFKA_NODE_ID: 2
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
KAFKA_LISTENERS: INTERNAL://kafka2:29092,CONTROLLER://kafka2:29093,EXTERNAL://0.0.0.0:9093
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka2:29092,EXTERNAL://localhost:9093
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka1:29093,2@kafka2:29093,3@kafka3:29093
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
KAFKA_LOG4J_ROOT_LOGLEVEL: WARN
KAFKA_AUTO_CREATE_TOPICS_ENABLE: false
CLUSTER_ID: p0KPFA_mQb2ixdPbQXPblw
healthcheck:
test: nc -zv kafka1 29092 || exit 1
interval: 5s
retries: 25
kafka3:
image: confluentinc/cp-server:7.5.0
hostname: kafka3
container_name: kafka3
ports:
- 9094:9094
environment:
KAFKA_NODE_ID: 3
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
KAFKA_LISTENERS: INTERNAL://kafka3:29092,CONTROLLER://kafka3:29093,EXTERNAL://0.0.0.0:9094
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka3:29092,EXTERNAL://localhost:9094
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka1:29093,2@kafka2:29093,3@kafka3:29093
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
KAFKA_LOG4J_ROOT_LOGLEVEL: WARN
KAFKA_AUTO_CREATE_TOPICS_ENABLE: false
CLUSTER_ID: p0KPFA_mQb2ixdPbQXPblw
healthcheck:
test: nc -zv kafka3 29092 || exit 1
interval: 5s
retries: 25
schema-registry:
image: confluentinc/cp-schema-registry:latest
hostname: schema-registry
container_name: schema-registry
ports:
- 8081:8081
environment:
SCHEMA_REGISTRY_HOST_NAME: schema-registry
SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka1:29092,kafka2:29092,kafka3:29092
SCHEMA_REGISTRY_LOG4J_ROOT_LOGLEVEL: WARN
SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
SCHEMA_REGISTRY_KAFKASTORE_TOPIC: _schemas
SCHEMA_REGISTRY_SCHEMA_REGISTRY_GROUP_ID: schema-registry
volumes:
- type: bind
source: .
target: /clientConfig
read_only: true
depends_on:
kafka1:
condition: service_healthy
kafka2:
condition: service_healthy
kafka3:
condition: service_healthy
healthcheck:
test: nc -zv schema-registry 8081 || exit 1
interval: 5s
retries: 25
gateway1:
image: harbor.cdkt.dev/conduktor/conduktor-gateway:3.5.0-SNAPSHOT
hostname: gateway1
container_name: gateway1
environment:
KAFKA_BOOTSTRAP_SERVERS: kafka1:29092,kafka2:29092,kafka3:29092
GATEWAY_ADVERTISED_HOST: localhost
GATEWAY_SECURITY_PROTOCOL: PLAINTEXT
GATEWAY_FEATURE_FLAGS_ANALYTICS: false
depends_on:
kafka1:
condition: service_healthy
kafka2:
condition: service_healthy
kafka3:
condition: service_healthy
ports:
- 6969:6969
- 6970:6970
- 6971:6971
- 6972:6972
- 8888:8888
healthcheck:
test: curl localhost:8888/health
interval: 5s
retries: 25
gateway2:
image: harbor.cdkt.dev/conduktor/conduktor-gateway:3.5.0-SNAPSHOT
hostname: gateway2
container_name: gateway2
environment:
KAFKA_BOOTSTRAP_SERVERS: kafka1:29092,kafka2:29092,kafka3:29092
GATEWAY_ADVERTISED_HOST: localhost
GATEWAY_SECURITY_PROTOCOL: PLAINTEXT
GATEWAY_FEATURE_FLAGS_ANALYTICS: false
depends_on:
kafka1:
condition: service_healthy
kafka2:
condition: service_healthy
kafka3:
condition: service_healthy
ports:
- 7969:6969
- 7970:6970
- 7971:6971
- 7972:6972
- 8889:8888
healthcheck:
test: curl localhost:8888/health
interval: 5s
retries: 25
kafka-client:
image: confluentinc/cp-kafka:latest
hostname: kafka-client
container_name: kafka-client
command: sleep infinity
volumes:
- type: bind
source: .
target: /clientConfig
read_only: true
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 berunning|healthy
. Implies detached mode.--detach
: Detached mode: Run containers in the background
- Command
- Output
- Recording
docker compose up --detach --wait
Network sql-topic-schema-registry_default Creating
Network sql-topic-schema-registry_default Created
Container kafka3 Creating
Container kafka1 Creating
Container kafka2 Creating
Container kafka-client Creating
Container kafka2 Created
Container kafka3 Created
Container kafka-client Created
Container kafka1 Created
Container gateway2 Creating
Container schema-registry Creating
Container gateway1 Creating
Container schema-registry Created
Container gateway1 Created
Container gateway2 Created
Container kafka-client Starting
Container kafka3 Starting
Container kafka2 Starting
Container kafka1 Starting
Container kafka3 Started
Container kafka1 Started
Container kafka-client Started
Container kafka2 Started
Container kafka3 Waiting
Container kafka1 Waiting
Container kafka3 Waiting
Container kafka1 Waiting
Container kafka2 Waiting
Container kafka3 Waiting
Container kafka2 Waiting
Container kafka1 Waiting
Container kafka2 Waiting
Container kafka3 Healthy
Container kafka1 Healthy
Container kafka2 Healthy
Container kafka3 Healthy
Container kafka1 Healthy
Container kafka3 Healthy
Container kafka2 Healthy
Container schema-registry Starting
Container kafka1 Healthy
Container gateway1 Starting
Container kafka2 Healthy
Container gateway2 Starting
Container gateway2 Started
Container schema-registry Started
Container gateway1 Started
Container schema-registry Waiting
Container gateway1 Waiting
Container gateway2 Waiting
Container kafka-client Waiting
Container kafka1 Waiting
Container kafka2 Waiting
Container kafka3 Waiting
Container kafka1 Healthy
Container kafka2 Healthy
Container kafka-client Healthy
Container kafka3 Healthy
Container schema-registry Healthy
Container gateway2 Healthy
Container gateway1 Healthy
Creating topic cars on gateway1
Creating on gateway1
:
- Topic
cars
with partitions:1 and replication-factor:1
- Command
- Output
- Recording
Listing topics in gateway1
- Command
- Output
- Recording
kafka-topics \
--bootstrap-server localhost:6969 \
--list
__consumer_offsets
_conduktor_gateway_acls
_conduktor_gateway_auditlogs
_conduktor_gateway_consumer_offsets
_conduktor_gateway_consumer_subscriptions
_conduktor_gateway_encryption_configs
_conduktor_gateway_groups
_conduktor_gateway_interceptor_configs
_conduktor_gateway_license
_conduktor_gateway_topicmappings
_conduktor_gateway_usermappings
_conduktor_gateway_vclusters
_confluent-command
_confluent-link-metadata
_schemas
cars
Produce avro payload
- Command
- Output
- Recording
schema='{
"type": "record",
"name": "car",
"fields": [
{"name": "type", "type": "string"},
{"name": "price", "type": "long"},
{"name": "color", "type": "string"}
]
}'
echo '{"type":"Sports","price":75,"color":"blue"}' | \
kafka-avro-console-producer \
--bootstrap-server localhost:6969 \
--topic cars \
--property schema.registry.url=http://localhost:8081 \
--property "value.schema=$schema"
echo '{"type":"SUV","price":55,"color":"red"}' | \
kafka-avro-console-producer \
--bootstrap-server localhost:6969 \
--topic cars \
--property schema.registry.url=http://localhost:8081 \
--property "value.schema=$schema"
[2024-11-26 10:26:34,034] INFO KafkaAvroSerializerConfig values:
auto.register.schemas = true
avro.reflection.allow.null = false
avro.remove.java.properties = false
avro.use.logical.type.converters = false
basic.auth.credentials.source = URL
basic.auth.user.info = [hidden]
bearer.auth.cache.expiry.buffer.seconds = 300
bearer.auth.client.id = null
bearer.auth.client.secret = null
bearer.auth.credentials.source = STATIC_TOKEN
bearer.auth.custom.provider.class = null
bearer.auth.identity.pool.id = null
bearer.auth.issuer.endpoint.url = null
bearer.auth.logical.cluster = null
bearer.auth.scope = null
bearer.auth.scope.claim.name = scope
bearer.auth.sub.claim.name = sub
bearer.auth.token = [hidden]
context.name.strategy = class io.confluent.kafka.serializers.context.NullContextNameStrategy
http.connect.timeout.ms = 60000
http.read.timeout.ms = 60000
id.compatibility.strict = true
key.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
latest.cache.size = 1000
latest.cache.ttl.sec = -1
latest.compatibility.strict = true
max.schemas.per.subject = 1000
normalize.schemas = false
proxy.host =
proxy.port = -1
rule.actions = []
rule.executors = []
rule.service.loader.enable = true
schema.format = null
schema.reflection = false
schema.registry.basic.auth.user.info = [hidden]
schema.registry.ssl.cipher.suites = null
schema.registry.ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
schema.registry.ssl.endpoint.identification.algorithm = https
schema.registry.ssl.engine.factory.class = null
schema.registry.ssl.key.password = null
schema.registry.ssl.keymanager.algorithm = SunX509
schema.registry.ssl.keystore.certificate.chain = null
schema.registry.ssl.keystore.key = null
schema.registry.ssl.keystore.location = null
schema.registry.ssl.keystore.password = null
schema.registry.ssl.keystore.type = JKS
schema.registry.ssl.protocol = TLSv1.3
schema.registry.ssl.provider = null
schema.registry.ssl.secure.random.implementation = null
schema.registry.ssl.trustmanager.algorithm = PKIX
schema.registry.ssl.truststore.certificates = null
schema.registry.ssl.truststore.location = null
schema.registry.ssl.truststore.password = null
schema.registry.ssl.truststore.type = JKS
schema.registry.url = [http://localhost:8081]
use.latest.version = false
use.latest.with.metadata = null
use.schema.id = -1
value.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
(io.confluent.kafka.serializers.KafkaAvroSerializerConfig:370)
[2024-11-26 10:26:36,644] INFO KafkaAvroSerializerConfig values:
auto.register.schemas = true
avro.reflection.allow.null = false
avro.remove.java.properties = false
avro.use.logical.type.converters = false
basic.auth.credentials.source = URL
basic.auth.user.info = [hidden]
bearer.auth.cache.expiry.buffer.seconds = 300
bearer.auth.client.id = null
bearer.auth.client.secret = null
bearer.auth.credentials.source = STATIC_TOKEN
bearer.auth.custom.provider.class = null
bearer.auth.identity.pool.id = null
bearer.auth.issuer.endpoint.url = null
bearer.auth.logical.cluster = null
bearer.auth.scope = null
bearer.auth.scope.claim.name = scope
bearer.auth.sub.claim.name = sub
bearer.auth.token = [hidden]
context.name.strategy = class io.confluent.kafka.serializers.context.NullContextNameStrategy
http.connect.timeout.ms = 60000
http.read.timeout.ms = 60000
id.compatibility.strict = true
key.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
latest.cache.size = 1000
latest.cache.ttl.sec = -1
latest.compatibility.strict = true
max.schemas.per.subject = 1000
normalize.schemas = false
proxy.host =
proxy.port = -1
rule.actions = []
rule.executors = []
rule.service.loader.enable = true
schema.format = null
schema.reflection = false
schema.registry.basic.auth.user.info = [hidden]
schema.registry.ssl.cipher.suites = null
schema.registry.ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
schema.registry.ssl.endpoint.identification.algorithm = https
schema.registry.ssl.engine.factory.class = null
schema.registry.ssl.key.password = null
schema.registry.ssl.keymanager.algorithm = SunX509
schema.registry.ssl.keystore.certificate.chain = null
schema.registry.ssl.keystore.key = null
schema.registry.ssl.keystore.location = null
schema.registry.ssl.keystore.password = null
schema.registry.ssl.keystore.type = JKS
schema.registry.ssl.protocol = TLSv1.3
schema.registry.ssl.provider = null
schema.registry.ssl.secure.random.implementation = null
schema.registry.ssl.trustmanager.algorithm = PKIX
schema.registry.ssl.truststore.certificates = null
schema.registry.ssl.truststore.location = null
schema.registry.ssl.truststore.password = null
schema.registry.ssl.truststore.type = JKS
schema.registry.url = [http://localhost:8081]
use.latest.version = false
use.latest.with.metadata = null
use.schema.id = -1
value.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
(io.confluent.kafka.serializers.KafkaAvroSerializerConfig:370)
Consume the avro payload back
- Command
- Output
- Recording
Creating topic red-cars on gateway1
Creating on gateway1
:
- Topic
red-cars
with partitions:1 and replication-factor:1
- Command
- Output
- Recording
Adding interceptor red-cars
Let's create the interceptor to filter out the red cars from the cars topic.
step-10-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 * FROM cars WHERE color = 'red'",
"schemaRegistryConfig" : {
"host" : "http://schema-registry:8081"
}
}
}
}
- Command
- Output
- Recording
curl \
--silent \
--request PUT "http://localhost:8888/gateway/v2/interceptor" \
--header "Content-Type: application/json" \
--user "admin:conduktor" \
--data @step-10-red-cars-interceptor.json | jq
{
"resource": {
"kind": "Interceptor",
"apiVersion": "gateway/v2",
"metadata": {
"name": "red-cars",
"scope": {
"vCluster": "passthrough",
"group": null,
"username": null
}
},
"spec": {
"comment": "Adding interceptor: red-cars",
"pluginClass": "io.conduktor.gateway.interceptor.VirtualSqlTopicPlugin",
"priority": 100,
"config": {
"virtualTopic": "red-cars",
"statement": "SELECT * FROM cars WHERE color = 'red'",
"schemaRegistryConfig": {
"host": "http://schema-registry:8081"
}
}
}
},
"upsertResult": "CREATED"
}
Listing interceptors
Listing interceptors on gateway1
- Command
- Output
- Recording
curl \
--silent \
--request GET "http://localhost:8888/gateway/v2/interceptor" \
--user "admin:conduktor" | jq
[
{
"kind": "Interceptor",
"apiVersion": "gateway/v2",
"metadata": {
"name": "red-cars",
"scope": {
"vCluster": "passthrough",
"group": null,
"username": null
}
},
"spec": {
"comment": "Adding interceptor: red-cars",
"pluginClass": "io.conduktor.gateway.interceptor.VirtualSqlTopicPlugin",
"priority": 100,
"config": {
"virtualTopic": "red-cars",
"statement": "SELECT * FROM cars WHERE color = 'red'",
"schemaRegistryConfig": {
"host": "http://schema-registry:8081"
}
}
}
}
]
- Command
- Output
- Recording
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.
- Command
- Output
- Recording
docker compose down --volumes
Container schema-registry Stopping
Container kafka-client Stopping
Container gateway2 Stopping
Container gateway1 Stopping
Container gateway2 Stopped
Container gateway2 Removing
Container gateway1 Stopped
Container gateway1 Removing
Container gateway2 Removed
Container schema-registry Stopped
Container schema-registry Removing
Container gateway1 Removed
Container schema-registry Removed
Container kafka1 Stopping
Container kafka3 Stopping
Container kafka2 Stopping
Container kafka3 Stopped
Container kafka3 Removing
Container kafka1 Stopped
Container kafka1 Removing
Container kafka3 Removed
Container kafka1 Removed
Container kafka-client Stopped
Container kafka-client Removing
Container kafka-client Removed
Container kafka2 Stopped
Container kafka2 Removing
Container kafka2 Removed
Network sql-topic-schema-registry_default Removing
Network sql-topic-schema-registry_default Removed
Conclusion
SQL virtual topics can also work on data serialized with Schema Registry!