Integrate Apache Kafka® with Hyper-Converged Database (HCD)

You can use Hyper-Converged Database (HCD) as a source of real-time change events for Apache Kafka®. The Kafka integration uses KafkaReplicator, a custom secondary index, to intercept row mutations at the coordinator and publish them to a Kafka topic. You don’t need to change your application’s write path.

With the Kafka integration, you can do the following:

Consume row mutations as JSON or Avro messages.
Enable event-driven pipelines, change-data-capture consumers, and analytics platforms.
Replicate to per-datacenter Kafka clusters in multi-datacenter deployments.

KafkaReplicator captures all write types in HCD, including the following:

Standard writes.
LOGGED batches (atomic across multiple partitions) and UNLOGGED batches (atomic only if they target a single partition). For more information, see BATCH.
Lightweight transactions (LWT/CAS) using IF NOT EXISTS and IF ... conditions. For more information, see Linearizable consistency.

Prerequisites

HCD 2.0.8 or later.
Host-level access to each HCD node that you want to replicate from, so that you can edit cassandra.yaml, set environment variables, and restart the node. For more information, see Kafka settings in cassandra.yaml and Set JVM system properties at startup.
A Kafka cluster running Confluent Platform 7.9 or later that is reachable from every HCD node that you want to replicate from.
A Kafka account that HCD can use to connect to your Kafka cluster, if the cluster requires authentication. The account must have permission to write to the destination topic. The account must also have permission to create topics if you set createTopicIfNotExists to true, and permission to read the events topic and use its consumer group if you enable deduplication.

For more information, see Security.
To use Avro format: A Confluent Schema Registry version 7.9 or later that is reachable from the same HCD nodes.
A keyspace and table in HCD that you want to replicate.

The table cannot have static columns or counter columns. For all restrictions, see Tables that cannot be indexed.

Set up the Kafka integration

To set up the Kafka integration, do the following:

Enable the Kafka integration and configure the connection to your Kafka cluster on each HCD node that you want to replicate from. Set the integration flag and the connection properties together, and then start or restart the node once.

In the examples, replace BROKER_1:9092,BROKER_2:9092 with the addresses of your Kafka brokers. The default is localhost:9092. To connect to a secured Kafka cluster, see Security. For all connection properties, see Configuration.

The method depends on your installation type:
Tarball installation
Pass --kafka and the connection properties to hcd cassandra:
bin/hcd cassandra --kafka \ -Dcassandra.kafka.bootstrap.servers=BROKER_1:9092,BROKER_2:9092
Debian or RHEL package installation
Add the connection properties to jvm-server.options:
-Dcassandra.kafka.bootstrap.servers=BROKER_1:9092,BROKER_2:9092
Then set HCD_KAFKA_ENABLED=1 in the service environment, and restart the service:
export HCD_KAFKA_ENABLED=1
Create a custom index on the table you want to replicate, using the KafkaReplicator class:
```
CREATE CUSTOM INDEX INDEX_NAME ON KEYSPACE_NAME.TABLE_NAME ()
USING 'KafkaReplicator'
WITH OPTIONS = {
    'topicName': 'TOPIC_NAME'
};
```
- INDEX_NAME: A name for the custom index.
- KEYSPACE_NAME.TABLE_NAME: The keyspace and table that you want to replicate.
- TOPIC_NAME: The Kafka topic to publish messages to.
To replicate only specific columns, list them in the parentheses after the table name:
```
CREATE CUSTOM INDEX INDEX_NAME ON KEYSPACE_NAME.TABLE_NAME (COLUMN_1, COLUMN_2)
USING 'KafkaReplicator'
WITH OPTIONS = {
    'topicName': 'TOPIC_NAME'
};
```
For a full list of options, see Index options.
Optional: To use Avro format, configure the Confluent Schema Registry URL. This is required when an index uses format=avro. Use any of the following methods:

System property

-Dcassandra.kafka.schema.registry.url=http://REGISTRY_HOST:8081

Environment variable

CASSANDRA_KAFKA_SCHEMA_REGISTRY_URL=http://REGISTRY_HOST:8081

cassandra.yaml

hcd.kafka.schema.registry.url: http://REGISTRY_HOST:8081

For more information, see Configuration.

The following example creates a products table, replicates it to a Kafka topic named ks.products, and creates the topic automatically if it doesn’t exist:

CREATE TABLE ks.products (
    id int PRIMARY KEY,
    name text,
    price float
);

CREATE CUSTOM INDEX products_kafka
ON ks.products (name, price)
USING 'KafkaReplicator'
WITH OPTIONS = {
    'topicName': 'ks.products',
    'createTopicIfNotExists': 'true'
};

For more examples, including Avro format, deduplication, and column projection, see Index examples.

Test the integration

To test the integration, do the following:

Write a row to the replicated table. For the previous example, run the following:
```
INSERT INTO ks.products (id, name, price) VALUES (42, 'Widget', 9.99);
```

Consume messages from the topic:

kafka-console-consumer \
  --bootstrap-server BROKER_HOST:9092 \
  --topic TOPIC_NAME \
  --from-beginning \
  --property print.key=true \
  --property print.headers=true

Replace the following:

BROKER_HOST: The address of a Kafka broker.
TOPIC_NAME: The topic you used in the index.

Verify that the message has the expected structure. For the previous example, the key is {"id":42}, the value is {"name":"Widget","price":9.99}, and the headers include rowVersion. For details, see Message format.
Optional: Delete the row and verify that a message with the same key and a null value (a tombstone) arrives on the topic:
```
DELETE FROM ks.products WHERE id = 42;
```

If messages don’t appear, see Troubleshoot.

Security

To connect to a secured Kafka cluster, pass the standard Kafka client security properties through the producer configuration. For more information, see Producer configuration.

Transparent Data Encryption

The replicator automatically applies the Transparent Data Encryption (TDE) settings from HCD to all files it writes to local storage for a given table.

After a key rotation, complete the following steps in order:

Run the key-rotation step (upgradesstables -a) described in the TDE documentation.
Restart the node or reload the index so the store picks up the new key. The store binds its cipher at open time, so you must complete this step before running checkpoint.
Run the checkpoint JMX operation. The replicator rewrites every pending record into fresh files under the current encryption configuration and deletes the older files.
Retire the previous key only after checkpoint() returns.

For more information, see About Transparent Data Encryption (TDE).

Troubleshoot

Use these troubleshooting tips to resolve common issues with the Kafka integration.

If mutations do not appear in Kafka, do the following:

Check the Kafka broker connection configuration and verify broker reachability.
Review logs for InProcessReplicatorService errors.
Check the KafkaMutationFailures and KafkaIndexerErrors metrics.
Verify that KafkaPendingOperations is not growing unboundedly, which could indicate a broker connectivity issue.

If you experience index build failures, do the following:

Verify that the Kafka broker is reachable from the HCD node.
Check that the topic exists or that you set createTopicIfNotExists to true.
Review logs for KafkaReplicator errors.
Check the KafkaIndexerInitialBuildFailures and KafkaIndexerRebuildFailures metrics.

If you experience high memory usage, storage capacity issues, or write throughput problems, do the following:

Reduce the max_pending_count configuration to limit memory pressure, or increase it to allow more pending mutations before backpressure applies.
Increase Kafka cluster capacity or add partitions to improve throughput.
Tune producer batching properties to increase Kafka indexing throughput.
Monitor the KafkaPendingOperations, KafkaAllowedMutations, and KafkaStorageMemtableBytes metrics.