Using CockroachDB CDC With Apache Pulsar
As of this writing, CockroachDB does not have a direct integration with Apache Pulsar. Here, integrate CockroachDB CDC with Pulsar via the existing Kafka sink.
Join the DZone community and get the full member experience.
Join For FreePrevious Articles on CockroachDB CDC
- Using CockroachDB CDC with Azure Event Hubs
- Using CockroachDB CDC with Confluent Cloud Kafka and Schema Registry
- SaaS Galore: Integrating CockroachDB with Confluent Kafka, Fivetran, and Snowflake
- CockroachDB CDC using Minio as a cloud storage sink
- CockroachDB CDC using Hadoop Ozone S3 Gateway as a cloud storage sink
Motivation
Apache Pulsar is a cloud-native distributed messaging and streaming platform. In my customer conversations, it most often comes up when compared to Apache Kafka. I have a customer needing a Pulsar sink support as they rely on Pulsar's multi-region capabilities. CockroachDB does not have a native Pulsar sink; however, the Pulsar project supports Kafka on Pulsar protocol support, and that's the core of today's article.
This tutorial assumes you have an enterprise license, you can also leverage our managed offerings where enterprise changefeeds are enabled by default. I am going to demonstrate the steps using a Docker environment instead.
High-Level Steps
- Deploy Apache Pulsar
- Deploy a CockroachDB cluster with enterprise changefeeds
- Deploy a Kafka Consumer
- Verify
- Conclusion
Step-By-Step Instructions
Deploy Apache Pulsar
Since I'm using Docker, I'm relying on the KoP Docker Compose environment provided by the Stream Native platform, which spearheads the development of Apache Pulsar.
I've used the service taken from the KoP example almost as-is aside from a few differences:
pulsar:
container_name: pulsar
hostname: pulsar
image: streamnative/sn-pulsar:2.11.0.5
command: >
bash -c "bin/apply-config-from-env.py conf/standalone.conf &&
exec bin/pulsar standalone -nss -nfw" # disable stream storage and functions worker
environment:
allowAutoTopicCreationType: partitioned
brokerDeleteInactiveTopicsEnabled: "false"
PULSAR_PREFIX_messagingProtocols: kafka
PULSAR_PREFIX_kafkaListeners: PLAINTEXT://pulsar:9092
PULSAR_PREFIX_brokerEntryMetadataInterceptors: org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptor
PULSAR_PREFIX_webServicePort: "8088"
ports:
- 6650:6650
- 8088:8088
- 9092:9092
I removed PULSAR_PREFIX_kafkaAdvertisedListeners: PLAINTEXT://127.0.0.1:19092
as I don't need it, I also changed the exposed port for - 19092:9092
to - 9092:9092
. My PULSAR_PREFIX_kafkaListeners
address points to a Docker container with the hostname pulsar
. I will need to access the address from other containers and I can't rely on the localhost. I'm also using a more recent version of the image than the one in their docs.
Deploy a CockroachDB Cluster With Enterprise Changefeeds
I am using a 3-node cluster in Docker. If you've followed my previous articles, you should be familiar with it.
I am using Flyway to set up the schema and seed the tables. The actual schema and data are taken from the changefeed examples we have in our docs. The only difference is I'm using a database called example
.
To enable CDC we need to execute the following commands:
SET CLUSTER SETTING cluster.organization = '<organization name>'; SET CLUSTER SETTING enterprise.license = '<secret>'; SET CLUSTER SETTING kv.rangefeed.enabled = true;
Again, if you don't have an enterprise license, you won't be able to complete this tutorial. Feel free to use our Dedicated or Serverless instances if you want to follow along.
Finally, after the tables and the data are in place, we can create a changefeed on these tables.
CREATE CHANGEFEED FOR TABLE office_dogs, employees INTO 'kafka://pulsar:9092';
Here I am using the Kafka port and the address of the Pulsar cluster, in my case pulsar
.
job_id ---------------------- 855538618543276035 (1 row) NOTICE: changefeed will emit to topic office_dogs NOTICE: changefeed will emit to topic employees Time: 50ms total (execution 49ms / network 1ms)
Everything seems to work and changefeed does not error out.
Deploy a Kafka Consumer
To validate data is being written to Pulsar, we need to stand up a Kafka client. I've created an image that downloads and installs Kafka. Once the entire Docker Compose environment is running, we can access the client and run the console consumer to verify.
/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server pulsar:9092 --topic office_dogs --from-beginning
{"after": {"id": 1, "name": "Petee H"}} {"after": {"id": 2, "name": "Carl"}}
If we want to validate that new data is flowing, let's insert another record into CockroachDB:
INSERT INTO office_dogs VALUES (3, 'Test');
The consumer will print a new row:
{"after": {"id": 3, "name": "Test"}}
Since we've created two topics, let's now look at the employees
topic.
/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server pulsar:9092 --topic employees --from-beginning
{"after": {"dog_id": 1, "employee_name": "Lauren", "rowid": 855539880336523267}} {"after": {"dog_id": 2, "employee_name": "Spencer", "rowid": 855539880336654339}}
Similarly, let's update a record and see the changes propagate to Pulsar.
UPDATE employees SET employee_name = 'Spencer Kimball' WHERE dog_id = 2;
{"after": {"dog_id": 2, "employee_name": "Spencer Kimball", "rowid": 855539880336654339}}
Verify
We've confirmed we can produce messages to Pulsar topics using the Kafka protocol via KoP. We've also confirmed we can consume using the Kafka console consumer. We can also use the native Pulsar tooling to confirm the data is consumable from Pulsar. I installed the Pulsar Python client, pip install pulsar-client
, on the Kafka client machine and created a Python script with the following code:
import pulsar client = pulsar.Client('pulsar://pulsar:6650') consumer = client.subscribe('employees', subscription_name='my-sub') while True: msg = consumer.receive() print("Received message: '%s'" % msg.data()) consumer.acknowledge(msg) client.close()
I execute the script:
root@kafka-client:/opt/kafka# python3 consume_messages.py 2023-04-11 14:17:21.761 INFO [281473255101472] Client:87 | Subscribing on Topic :employees 2023-04-11 14:17:21.762 INFO [281473255101472] ClientConnection:190 | [<none> -> pulsar://pulsar:6650] Create ClientConnection, timeout=10000 2023-04-11 14:17:21.762 INFO [281473255101472] ConnectionPool:97 | Created connection for pulsar://pulsar:6650 2023-04-11 14:17:21.763 INFO [281473230237984] ClientConnection:388 | [172.28.0.3:33826 -> 172.28.0.6:6650] Connected to broker 2023-04-11 14:17:21.771 INFO [281473230237984] HandlerBase:72 | [persistent://public/default/employees-partition-0, my-sub, 0] Getting connection from pool 2023-04-11 14:17:21.776 INFO [281473230237984] ClientConnection:190 | [<none> -> pulsar://pulsar:6650] Create ClientConnection, timeout=10000 2023-04-11 14:17:21.776 INFO [281473230237984] ConnectionPool:97 | Created connection for pulsar://localhost:6650 2023-04-11 14:17:21.776 INFO [281473230237984] ClientConnection:390 | [172.28.0.3:33832 -> 172.28.0.6:6650] Connected to broker through proxy. Logical broker: pulsar://localhost:6650 2023-04-11 14:17:21.786 INFO [281473230237984] ConsumerImpl:238 | [persistent://public/default/employees-partition-0, my-sub, 0] Created consumer on broker [172.28.0.3:33832 -> 172.28.0.6:6650] 2023-04-11 14:17:21.786 INFO [281473230237984] MultiTopicsConsumerImpl:274 | Successfully Subscribed to a single partition of topic in TopicsConsumer. Partitions need to create : 0 2023-04-11 14:17:21.786 INFO [281473230237984] MultiTopicsConsumerImpl:137 | Successfully Subscribed to Topics
Let's insert a record into the employees
tables:
INSERT INTO employees (dog_id, employee_name) VALUES (3, 'Test'); UPDATE employees SET employee_name = 'Artem' WHERE dog_id = 3;
The Pulsar client output is as follows:
Received message: 'b'{"after": {"dog_id": 3, "employee_name": "Test", "rowid": 855745376561364994}}'' Received message: 'b'{"after": {"dog_id": 3, "employee_name": "Artem", "rowid": 855745376561364994}}''
Conclusion
This is how you can leverage existing CockroachDB capability with non-standard services like Apache Pulsar. Hopefully, you've found this article useful and can start leveraging the existing Kafka sink with non-standard message brokers.
Published at DZone with permission of Artem Ervits. See the original article here.
Opinions expressed by DZone contributors are their own.
Comments