Apache Kafka® integration reference

Use these options and properties to configure the Kafka integration, including message format, Avro schema mapping, index options, examples, and configuration settings. For setup instructions, see Integrate Apache Kafka® with Hyper-Converged Database (HCD).

Supported versions

The HCD Kafka integration is tested and verified with the following versions:

Component Supported version Notes

Confluent Platform

7.9 or later

Maps to Apache Kafka® 3.9.x.

Confluent Schema Registry

7.9 or later

Required when using format=avro.

Message format

Every HCD mutation produces one Kafka message. The format index option (json or avro) controls the serialization format. Both formats share the same structure for key, value, and headers.

The message key contains all primary key columns (partition key and clustering columns).

  • JSON: A JSON object. For example, {"id": 42} or {"tenant": "acme", "id": 42}.

  • Avro: A GenericRecord whose schema covers the same primary-key columns.

The message value contains the non-primary-key column values for an INSERT or UPDATE, or null (a Kafka tombstone) for a DELETE.

  • JSON: A JSON object. For example, {"name": "Widget", "price": 9.99}, or null for DELETE.

  • Avro: A GenericRecord (all value fields are nullable unions with default null), or null for DELETE.

When you configure a column projection, only the projected columns appear in the value. Primary-key columns are always included in the key regardless of projection.

Headers are UTF-8 string values attached to every record regardless of format:

  • rowVersion: The row write-timestamp as a decimal string (microseconds since Unix epoch). For example, "1718000000000000". Present on most live writes. Corresponds to WRITETIME(). Use it for last-write-wins semantics or reordering detection.

  • late: Set to "true" when the mutation arrived out of order (for example, a late or compensating write). Normal, in-order inserts and updates do not carry this header.

If your application cannot handle the rowVersion and late headers, enable deduplication.

Avro format and Schema Registry

When format=avro is set on the index, HCD does the following:

  • Derives schemas automatically from the HCD table metadata at index-creation time.

  • Registers schemas in the Confluent Schema Registry under the standard Confluent subjects: <topic>-key and <topic>-value.

  • Serializes both key and value using KafkaAvroSerializer.

  • For a DELETE, sends an Avro-encoded key with a null value (tombstone).

format=avro is not compatible with unpackJsonFields. Creating an index with both options set fails with a validation error.

CQL to Avro type mapping

The schema builder maps every HCD CQL type to a semantically equivalent Avro type, as shown in the following table:

CQL type Avro base type Avro logical type Notes

tinyint, smallint, int

INT

N/A

bigint, counter

LONG

N/A

float

FLOAT

N/A

double

DOUBLE

N/A

boolean

BOOLEAN

N/A

blob

BYTES

N/A

text, varchar, ascii, inet

STRING

N/A

uuid, timeuuid

STRING

uuid

RFC-4122 UUID string.

timestamp

LONG

timestamp-millis

Milliseconds since Unix epoch.

date

INT

date

Days since Unix epoch.

time

LONG

time-micros

Microseconds since midnight.

varint

BYTES

cql_varint

Big-endian two’s-complement bytes.

decimal

RECORD {bigint: BYTES, scale: INT}

cql_decimal

Unscaled value and scale.

duration

RECORD {months: INT, days: INT, nanoseconds: LONG}

cql_duration

Three-component CQL duration.

list<T>, set<T>

ARRAY of T

N/A

Element type mapped recursively. Frozen and non-frozen both produce ARRAY.

map<K,V>

MAP of V

N/A

Keys always stringified to STRING. Value type mapped recursively.

tuple<T0,T1,…>

RECORD {_0, _1, …}

N/A

Positional fields named _0, _1, and so on. Each field is a nullable union ["null", T].

UDT

RECORD with UDT field names

N/A

Record name = UDT name. Namespace = UDT keyspace. Each field is a nullable union.

vector<T,N>

ARRAY of T

N/A

HCD only. Fixed-length array of N elements.

point, linestring, polygon

STRING

N/A

HCD-only geospatial types. No Avro equivalent. Logs WARNING. Value serialized as WKT.

date_range

STRING

N/A

HCD-only type. No Avro equivalent. Logs WARNING.

All non-primary-key value fields are wrapped as nullable unions ["null", <type>] with default null to accommodate HCD’s sparse row model and partial updates.

Index options

Pass all options as string values inside WITH OPTIONS = { …​ }.

The fully qualified class name for the KafkaReplicator custom index is com.datastax.kafka.KafkaReplicator. When the implementation is registered as a trusted secondary index, use the short form KafkaReplicator directly.

Topic and lifecycle

The options that control the Kafka topic and initial build behavior of the index:

Option Default Description

topicName

hcd_<keyspace>_<table>

The Kafka topic to produce messages to. Must be non-blank if supplied.

createTopicIfNotExists

false

When true, HCD creates the Kafka topic on CREATE INDEX if it does not already exist. When deduplication is enabled, HCD also creates the events topic.

partitions

1

Number of partitions when auto-creating the topic. Also the default for eventsTopicPartitions. Must be a positive integer.

replicationFactor

1

Replication factor when auto-creating the topic. Also the default for eventsTopicReplicationFactor. Must be a positive short.

skipBuildIfTableNotEmpty

false

When true, HCD skips the initial SSTable scan even when the table already contains data. HCD replicates only live writes made after CREATE INDEX. Useful when you want to backfill separately or when existing data is not needed in Kafka.

Deduplication

These options control the optional deduplication pipeline. For more information, see Deduplication.

The deduplication options:

Option Default Description

deduplicationEnabled

false

When true, enables automatic versions deduplication through an intermediate events topic and a background EventsConsumerWorker. When false, HCD uses standard direct replication.

eventsTopic

<topicName>-events

Name of the intermediate events topic used when deduplicationEnabled=true.

eventsTopicPartitions

Value of partitions

Number of partitions for the events topic when HCD auto-creates it.

eventsTopicReplicationFactor

Value of replicationFactor

Replication factor for the events topic when HCD auto-creates it.

eventsConsumerGroupId

hcd-dedup-<keyspace>-<table>

Kafka consumer group ID shared by all EventsConsumerWorker instances for this table across the cluster.

Serialization format

The option that controls the serialization format of Kafka messages:

Option Default Description

format

json

Serialization format for Kafka messages. Accepted values: json, avro. avro requires a configured Schema Registry URL. Incompatible with unpackJsonFields.

Column projection

These options control which columns are replicated in the message value. Primary-key columns are always included in the message key.

The column projection options:

Option Default Description

target

(none)

Columns specified between the parentheses in CREATE CUSTOM INDEX …​ ON table (col1, col2). CQL populates this automatically. Do not set it explicitly inside WITH OPTIONS.

columnsToIndex

(all non-PK columns)

Comma-separated list of non-primary-key columns to include in the message value. Omit to include every column.

Non-primary replica processing

The option that controls whether non-primary HCD replicas participate in replication:

Option Default Description

indexNonPrimaryReplicas

true

When true, every HCD replica independently processes each mutation. Set to false so that only the primary replica processes rows, avoiding duplicate processing. With false, HCD does not replicate rows held exclusively by a non-primary replica until a rebuild or until that replica becomes primary.

Tables that cannot be indexed

CREATE CUSTOM INDEX raises InvalidRequestException for the following table types:

  • Tables with static columns.

  • Tables with counter columns.

Index examples

The following examples show common CREATE CUSTOM INDEX configurations for different deployment scenarios.

Minimal JSON setup

Create a basic index with an explicit topic name. This example assumes that you enabled the Kafka integration and configured the broker connection as described in Integrate Apache Kafka® with Hyper-Converged Database (HCD):

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'};

-- INSERT: key: {"id":42}  value: {"name":"Widget","price":9.99}
INSERT INTO ks.products (id, name, price) VALUES (42, 'Widget', 9.99);

-- DELETE: key: {"id":42}  value: null  (tombstone)
DELETE FROM ks.products WHERE id = 42;

Avro format with Schema Registry

Use Avro serialization to automatically derive and register schemas in the Confluent Schema Registry:

-- Additional system property required:
--   -Dcassandra.kafka.schema.registry.url=http://localhost:8081

CREATE CUSTOM INDEX products_avro
    ON ks.products (name, price)
    USING 'KafkaReplicator'
    WITH OPTIONS = {
        'topicName': 'ks.products',
        'format':    'avro'
    };
-- Schemas are automatically registered as subjects
-- "ks.products-key" and "ks.products-value".

Auto-create topic with multiple partitions

Create the Kafka topic automatically at index-creation time with custom partition and replication settings:

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

Index only selected columns (projection)

Replicate only a subset of non-primary-key columns to reduce message size:

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

Deduplication mode with an intermediate events topic

Enable deduplication so that HCD publishes only the latest version of each row to the destination topic. HCD creates both the destination topic and the events topic if they do not exist:

CREATE CUSTOM INDEX products_dedup
    ON ks.products (name, price)
    USING 'KafkaReplicator'
    WITH OPTIONS = {
        'topicName':              'ks.products',
        'deduplicationEnabled':   'true',
        'eventsTopic':            'ks.products-events',
        'eventsTopicPartitions':  '6',
        'eventsConsumerGroupId':  'hcd-dedup-ks-products',
        'createTopicIfNotExists': 'true'
    };

Skip initial backfill

Index only writes made after CREATE INDEX, leaving pre-existing rows out of Kafka until a separate backfill is performed:

CREATE CUSTOM INDEX events_kafka
    ON analytics.events ()
    USING 'KafkaReplicator'
    WITH OPTIONS = {
        'topicName':                'analytics.events',
        'skipBuildIfTableNotEmpty': 'true'
    };

Configuration

You can provide settings as JVM system properties, environment variables, or cassandra.yaml keys.

When HCD is started with HCDConfigurationLoader (the default), you can place all Kafka settings in the hcd: top-level section of cassandra.yaml. For an annotated cassandra.yaml example covering every available setting, see Kafka settings in cassandra.yaml.

System properties take precedence over environment variables, which take precedence over cassandra.yaml values, which take precedence over built-in defaults.

Kafka broker connection

The settings that configure the connection to the Kafka broker:

Property Environment variable Default Description

cassandra.kafka.bootstrap.servers

CASSANDRA_KAFKA_BOOTSTRAP_SERVERS

localhost:9092

Comma-separated list of Kafka broker addresses.

cassandra.kafka.schema.registry.url

CASSANDRA_KAFKA_SCHEMA_REGISTRY_URL

(none)

Confluent Schema Registry URL. Required when any index uses format=avro.

Producer configuration

HCD creates the producer with sensible batching defaults that you can override through the three-level hierarchy.

The built-in producer defaults:

Producer property Default Purpose

batch.size

65536 (64 KB)

Increased from the Kafka default of 16 KB.

linger.ms

1

Wait up to 1 ms for more messages to accumulate.

compression.type

lz4

Compression to maximize batch efficiency.

buffer.memory

33554432 (32 MB)

Total in-memory buffer for the producer.

The mandatory producer properties, which you cannot override:

Producer property JSON mode Avro mode

bootstrap.servers

Configured value

Configured value

key.serializer

StringSerializer

KafkaAvroSerializer

value.serializer

StringSerializer

KafkaAvroSerializer

schema.registry.url

Not set

Configured Schema Registry URL

To supply extra producer properties, use one of the following methods:

  • cassandra.yaml: hcd.kafka.producer map (flat key-value pairs)

  • Environment variable: CASSANDRA_KAFKA_PRODUCER_<UPPER_SNAKE>, lowercased with dots

  • System property: cassandra.kafka.producer.<property.name>

The following is an example cassandra.yaml with producer configuration:

hcd:
  kafka:
    bootstrap:
      servers: broker1:9092,broker2:9092
    schema:
      registry:
        url: http://registry:8081
    producer:
      acks: "all"
      retries: 3
      max.block.ms: 5000

Values must be strings, booleans, or numbers. HCD rejects lists and maps at startup.

Consumer configuration

HCD creates a Kafka consumer and a background EventsConsumerWorker only when an index has deduplicationEnabled: true. In standard mode, no Kafka consumer is started.

The consumer uses the following built-in defaults:

Consumer property Default Purpose

enable.auto.commit

false

Mandatory. HCD commits offsets explicitly and asynchronously after each batch is processed.

auto.offset.reset

earliest

Ensures no events are skipped when a worker joins a consumer group for the first time.

key.deserializer

ByteArrayDeserializer

Binary deserializer for the serialized primary key.

value.deserializer

ByteArrayDeserializer

Binary deserializer for the events topic message value.

group.id

hcd-dedup-<keyspace>-<table>

Overridden per table by the eventsConsumerGroupId index option.

The properties that tune consumer threading and polling:

Property Environment variable Default Description

cassandra.kafka.consumer.threads

KAFKA_CONSUMER_THREADS

4

Number of consumer threads in the shared consumer pool for Kafka event polling.

cassandra.kafka.consumer.worker.threads

KAFKA_CONSUMER_WORKER_THREADS

4

Number of worker threads per table pool for concurrent HCD reads.

cassandra.kafka.consumer.pool.timeout.millis

KAFKA_CONSUMER_POOL_TIMEOUT_MILLIS

1000 (1 s)

Timeout in milliseconds for consumer.poll() before the worker completes a batch.

To set extra consumer properties, use one of the following methods:

  • cassandra.yaml: hcd.kafka.consumer map (flat key-value pairs)

  • Environment variable: CASSANDRA_KAFKA_CONSUMER_<UPPER_SNAKE>, lowercased with dots

  • System property: cassandra.kafka.consumer.<property.name>

Values must be strings, booleans, or numbers. HCD rejects lists and maps at startup.

Admin client configuration

HCD creates an Admin client only when createTopicIfNotExists: true is set on an index. To set admin client configuration, use one of the following methods:

  • cassandra.yaml: hcd.kafka.admin map (flat key-value pairs)

  • Environment variable: CASSANDRA_KAFKA_ADMIN_<UPPER_SNAKE>, lowercased with dots

  • System property: cassandra.kafka.admin.<property.name>

Mutation processing

The properties that tune the internal mutation processing pipeline:

Property Environment variable Default Description

cassandra.kafka.processing.mutation_timeout

CASSANDRA_KAFKA_PROCESSING_MUTATION_TIMEOUT

60s

Timeout for mutation processing.

cassandra.kafka.processing.mutation_process_threads

CASSANDRA_KAFKA_PROCESSING_MUTATION_PROCESS_THREADS

16

Number of threads for mutation processing.

cassandra.kafka.processing.query_threads

CASSANDRA_KAFKA_PROCESSING_QUERY_THREADS

16

Number of threads for HCD read-back queries.

cassandra.kafka.processing.remediation_timeout

CASSANDRA_KAFKA_PROCESSING_REMEDIATION_TIMEOUT

1s

Timeout for remediation operations, which are retries that attempt to recover a mutation that failed initial processing.

cassandra.kafka.processing.datacenter_aware_replication

CASSANDRA_KAFKA_PROCESSING_DATACENTER_AWARE_REPLICATION

true

When true, each datacenter writes only to its local Kafka cluster. For more information, see Multi-datacenter replication.

cassandra.kafka.processing.retry.max_retries

CASSANDRA_KAFKA_PROCESSING_RETRY_MAX_RETRIES

unbounded

Maximum retry attempts for general processing operations.

cassandra.kafka.processing.retry.initial_interval

CASSANDRA_KAFKA_PROCESSING_RETRY_INITIAL_INTERVAL

1s

Initial retry delay for general processing operations.

cassandra.kafka.processing.retry.max_interval

CASSANDRA_KAFKA_PROCESSING_RETRY_MAX_INTERVAL

30s

Maximum retry delay for general processing operations (exponential backoff cap).

Storage

The replicator writes pending mutations to a durable write-ahead log on disk, surviving crashes and node restarts.

The storage properties:

Property Environment variable Default Description

cassandra.kafka.storage.directory

CASSANDRA_KAFKA_STORAGE_DIRECTORY

(auto-discovered)

Directory for persistent storage. When empty, HCD auto-discovers the directory from Cassandra’s data directory configuration. Typically CASSANDRA_DATA_DIR/kafka/.

cassandra.kafka.storage.max_pending_count

CASSANDRA_KAFKA_STORAGE_MAX_PENDING_COUNT

100000

Maximum number of pending mutations before HCD applies backpressure and slows incoming writes.

The disk storage properties:

Property Environment variable Default Description

cassandra.kafka.storage.memtable_byte_budget

CASSANDRA_KAFKA_STORAGE_MEMTABLE_BYTE_BUDGET

536870912 (512 MiB)

Per-table memtable size limit.

cassandra.kafka.storage.commit_log_max_file_bytes

CASSANDRA_KAFKA_STORAGE_COMMIT_LOG_MAX_FILE_BYTES

8388608 (8 MiB)

Maximum commit log file size.

cassandra.kafka.storage.commit_log_checkpoint_threshold_bytes

CASSANDRA_KAFKA_STORAGE_COMMIT_LOG_CHECKPOINT_THRESHOLD_BYTES

33554432 (32 MiB)

Commit-log size that triggers a checkpoint, after which the replicator compacts and truncates the log.

cassandra.kafka.storage.force_max_wait_nanos

CASSANDRA_KAFKA_STORAGE_FORCE_MAX_WAIT_NANOS

1000000 (1 ms)

Maximum time in nanoseconds the replicator waits to batch multiple fsyncs into a single fdatasync call.

cassandra.kafka.storage.flush_when_queue_empty

CASSANDRA_KAFKA_STORAGE_FLUSH_WHEN_QUEUE_EMPTY

true

Flush immediately when the queue is empty (low-latency mode).

cassandra.kafka.storage.shared_memtable_budget

CASSANDRA_KAFKA_STORAGE_SHARED_MEMTABLE_BUDGET

1073741824 (1 GiB)

Shared memory budget across all tables.

cassandra.kafka.storage.complete_on_write

CASSANDRA_KAFKA_STORAGE_COMPLETE_ON_WRITE

true

Complete startMutation futures on write to the OS page cache rather than waiting for fsync.

cassandra.kafka.storage.completed_mutations_flush_interval_ms

CASSANDRA_KAFKA_STORAGE_COMPLETED_MUTATIONS_FLUSH_INTERVAL_MS

1

Flush interval in milliseconds for completed mutations.

The replicator maintains persistent state in the following directory structure, where <storage.directory> is the configured storage directory, <keyspace> and <table> identify the HCD table being replicated, and <id> is a file identifier the replicator assigns automatically:

<storage.directory>/
├── <keyspace>/
│   └── <table>/
│       ├── manifest.json     # Current checkpoint state
│       ├── cl-<id>.log       # Commit log files (write-ahead log)
│       └── seg-<id>.dat      # Data segment files (flushed memtables)

  • Commit log: Rolling log files (cl-*.log) that record all pending mutations for crash recovery.

  • Segments: Immutable data files (seg-*.dat) containing data written from memory to disk.

  • Manifest: JSON file tracking current checkpoint state and active files.

Was this helpful?

Give Feedback

How can we improve the documentation?

© Copyright IBM Corporation 2026 | Privacy policy | Terms of use |  Manage Privacy Choices

Apache, Apache Cassandra, Cassandra, Apache Tomcat, Tomcat, Apache Lucene, Apache Solr, Apache Hadoop, Hadoop, Apache Pulsar, Pulsar, Apache Spark, Spark, Apache TinkerPop, TinkerPop, Apache Kafka and Kafka are either registered trademarks or trademarks of the Apache Software Foundation or its subsidiaries in Canada, the United States and/or other countries. Kubernetes is the registered trademark of the Linux Foundation.

General Inquiries: Contact IBM