feat: CDC-based mixed index synchronization (#4873) by porunov · Pull Request #4906 · JanusGraph/janusgraph

porunov · 2026-06-28T10:01:55Z

Keep mixed indexes (ElasticSearch/Solr/Lucene) eventually consistent with the graph by deriving their updates from a Change-Data-Capture stream of the committed graph data, instead of a synchronous second write during the transaction that can diverge on failure and leave a permanently stale index.

Pipeline (Apache Cassandra): commit (graph data only) -> Cassandra edgestore(cdc=true) ->
Debezium -> Kafka -> CdcIndexUpdateWorker consumer group -> reindex-from-current-state -> ES bulk.

Key design points:

Reindex-from-current-state: the worker reads each changed element's current graph state and fully replaces its index document (reusing IndexSerializer, like transaction recovery). This is idempotent and order-independent, so out-of-order or duplicate events still converge to the current state -- no strict ordering required, and a stale event can never overwrite a fresh value.
No dual write: the only synchronous write is to storage; the index is updated downstream from the committed change stream, so it cannot diverge.
Element-keyed Kafka partitioning gives per-element ordering and horizontal scaling via a consumer group; batches are de-duplicated and applied as one ElasticSearch _bulk per index.
At-least-once: offsets are committed only after a batch is durably applied; on failure the batch is reprocessed (rewind) rather than skipped, so the index eventually catches up.

Configuration (opt-in, disabled by default):

storage.cql.cdc: emit the Cassandra cdc=true table option on the edgestore table.
index.[X].cdc.enabled / index.[X].cdc.synchronous: per-index dual mode (write synchronously AND via CDC) or cdc-only mode (skip the synchronous write; ES updated solely via CDC).

Components:

janusgraph-core: per-index CDC config options, the commit-side skip hook in StandardJanusGraph, and MixedIndexUpdateApplier (the backend-agnostic reindex-from-current-state engine).
janusgraph-cql: the storage.cql.cdc table option (no Kafka dependency in production code).
janusgraph-cdc (new module; core + kafka-clients): the CdcEventDecoder SPI, DebeziumCassandraJsonDecoder, CdcWorkerConfiguration, CdcIndexUpdateWorker, and the standalone CdcIndexUpdateWorkerMain runner.

Testing: 38 tests, including unit/component coverage (decoder vs real serialized bytes, reindex engine, worker loop via Kafka MockConsumer, full-chain convergence over Lucene incl. vertex/edge add/update/remove and out-of-order delivery) and two real-container E2Es -- worker -> Kafka -> ElasticSearch, and the full Cassandra-CDC -> Debezium -> Kafka -> ElasticSearch pipeline. The full Debezium E2E is gated behind the cassandra-cdc-e2e Maven profile (auto-activated on Java 17+, required by Debezium 3.x and Testcontainers 2.x); the default Java 8/11 build excludes it and stays green.

Docs: advanced-topics/cdc-mixed-index.md operator guide, a 1.2.0 changelog upgrade note, and the regenerated configuration reference.

Fixes #4873
Replaces #4874

Thank you for contributing to JanusGraph!

In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:

For all changes:

Is there an issue associated with this PR? Is it referenced in the commit message?
Does your PR body contain #xyz where xyz is the issue number you are trying to resolve?
Has your PR been rebased against the latest commit within the target branch (typically master)?
Is your initial contribution a single, squashed commit?

For code changes:

Have you written and/or updated unit tests to verify your changes?
If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under ASF 2.0?
If applicable, have you updated the LICENSE.txt file, including the main LICENSE.txt file in the root of this repository?
If applicable, have you updated the NOTICE.txt file, including the main NOTICE.txt file found in the root of this repository?

For documentation related changes:

Have you ensured that format looks appropriate for the output in which it is rendered?

Copilot

Pull request overview

Adds an opt-in Change-Data-Capture (CDC) pipeline to keep mixed indexes (ElasticSearch/Solr/Lucene) eventually consistent with committed graph data by asynchronously reindexing affected elements from their current state, eliminating the “dual write” divergence risk.

Changes:

Introduces per-index CDC configuration (index.[X].cdc.*) and commit-path logic to skip synchronous mixed-index writes in cdc-only mode.
Adds MixedIndexUpdateApplier (reindex-from-current-state) and CdcElementChange in core, plus Cassandra storage.cql.cdc table option support.
Adds a new janusgraph-cdc module implementing a Kafka consumer worker, Debezium Cassandra JSON decoder, and extensive unit/component/E2E tests + documentation.

Reviewed changes

Copilot reviewed 33 out of 33 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
pom.xml	Registers new `janusgraph-cdc` Maven module in the build.
mkdocs.yml	Adds CDC operator guide page to documentation nav.
janusgraph-lucene/src/test/java/org/janusgraph/diskstorage/lucene/MixedIndexUpdateApplierTest.java	Validates reindex-from-current-state behavior over Lucene in cdc-only mode.
janusgraph-lucene/src/test/java/org/janusgraph/diskstorage/lucene/CdcSkipMutationTest.java	Verifies synchronous mixed-index write is skipped in cdc-only and retained in dual mode.
janusgraph-cql/src/test/java/org/janusgraph/diskstorage/cql/CQLCdcTableOptionTest.java	Unit-tests that `storage.cql.cdc` toggles `cdc=true` on `edgestore` DDL only.
janusgraph-cql/src/main/java/org/janusgraph/diskstorage/cql/CQLKeyColumnValueStore.java	Refactors CREATE TABLE building and conditionally applies Cassandra `cdc=true` for `edgestore`.
janusgraph-cql/src/main/java/org/janusgraph/diskstorage/cql/CQLConfigOptions.java	Adds `storage.cql.cdc` configuration option.
janusgraph-core/src/test/java/org/janusgraph/graphdb/configuration/CdcIndexConfigTest.java	Tests defaults and per-index scoping of `index.[X].cdc.*`.
janusgraph-core/src/main/java/org/janusgraph/graphdb/database/StandardJanusGraph.java	Computes cdc-only backing indexes and skips synchronous mixed-index writes for them.
janusgraph-core/src/main/java/org/janusgraph/graphdb/database/log/TransactionLogHeader.java	Makes `TransactionLogHeader.Modification` constructor public for reuse in decoding.
janusgraph-core/src/main/java/org/janusgraph/graphdb/database/index/MixedIndexUpdateApplier.java	Adds backend-agnostic reindex-from-current-state applier for CDC worker.
janusgraph-core/src/main/java/org/janusgraph/graphdb/database/index/CdcElementChange.java	Adds normalized “element changed” model consumed by the applier/worker.
janusgraph-core/src/main/java/org/janusgraph/graphdb/configuration/GraphDatabaseConfiguration.java	Adds `index.[X].cdc.enabled` and `index.[X].cdc.synchronous` options.
janusgraph-cdc/src/test/resources/cassandra-cdc.yaml	Provides Cassandra config enabling CDC for full pipeline E2E test.
janusgraph-cdc/src/test/java/org/janusgraph/cdc/DebeziumCassandraJsonDecoderTest.java	Tests Debezium JSON decoding against real JanusGraph-serialized bytes.
janusgraph-cdc/src/test/java/org/janusgraph/cdc/CdcWorkerConvergenceTest.java	Drives worker+decoder+applier via `MockConsumer` to validate convergence semantics.
janusgraph-cdc/src/test/java/org/janusgraph/cdc/CdcWorkerConfigurationTest.java	Tests worker configuration defaults and properties parsing.
janusgraph-cdc/src/test/java/org/janusgraph/cdc/CdcKafkaElasticsearchTest.java	Testcontainers E2E for Kafka → worker → ElasticSearch convergence.
janusgraph-cdc/src/test/java/org/janusgraph/cdc/CdcIndexUpdateWorkerMainTest.java	Tests runner wiring and config reflection of CDC-enabled backing indexes.
janusgraph-cdc/src/test/java/org/janusgraph/cdc/CdcIndexUpdateWorkerLoopTest.java	Unit-tests polling loop semantics (dedupe/retry/commit/rewind).
janusgraph-cdc/src/test/java/org/janusgraph/cdc/CdcEventDecoderTest.java	Smoke test for decoder SPI and `CdcElementChange` interop.
janusgraph-cdc/src/test/java/org/janusgraph/cdc/CdcCassandraDebeziumElasticsearchTest.java	Full Cassandra CDC → Debezium → Kafka → ElasticSearch pipeline E2E (profile-gated).
janusgraph-cdc/src/test/java/io/debezium/connector/cassandra/JanusGraphCdcConnectorStarter.java	Test-only bridge to start Debezium Cassandra connector embedded.
janusgraph-cdc/src/main/java/org/janusgraph/cdc/DebeziumCassandraJsonDecoder.java	Implements Debezium Cassandra JSON → `CdcElementChange` decoding.
janusgraph-cdc/src/main/java/org/janusgraph/cdc/CdcWorkerConfiguration.java	Defines immutable worker/Kafka configuration and consumer properties.
janusgraph-cdc/src/main/java/org/janusgraph/cdc/CdcIndexUpdateWorkerMain.java	Adds standalone runner that opens graph, wires decoder+applier, starts workers.
janusgraph-cdc/src/main/java/org/janusgraph/cdc/CdcIndexUpdateWorker.java	Implements Kafka consume/decode/dedupe/apply/retry/commit/rewind loop.
janusgraph-cdc/src/main/java/org/janusgraph/cdc/CdcIndexApplier.java	Functional interface to abstract index application for worker tests.
janusgraph-cdc/src/main/java/org/janusgraph/cdc/CdcEventDecoder.java	Decoder SPI for CDC record formats.
janusgraph-cdc/pom.xml	New module POM with Kafka clients + testcontainers/Debezium profile gating.
docs/configs/janusgraph-cfg.md	Regenerates config reference including new CDC options.
docs/changelog.md	Adds 1.2.0 upgrade note describing CDC mixed index synchronization.
docs/advanced-topics/cdc-mixed-index.md	Adds operator guide for Cassandra CDC + Debezium + Kafka + worker setup.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot

Pull request overview

Copilot reviewed 35 out of 35 changed files in this pull request and generated 3 comments.

Copilot

Pull request overview

Copilot reviewed 35 out of 35 changed files in this pull request and generated 4 comments.

Comments suppressed due to low confidence (1)

janusgraph-lucene/src/test/java/org/janusgraph/diskstorage/lucene/MixedIndexUpdateApplierTest.java:1

The 10-second schema/index enablement timeout is likely to be flaky under loaded CI runners (especially with filesystem-backed Lucene + BerkeleyJE). Consider increasing these timeouts (e.g., 30–60 seconds) or using a shared constant aligned with other JanusGraph index-status tests. The same concern applies to other new tests using 10-second awaitGraphIndexStatus timeouts.

Copilot

Pull request overview

Copilot reviewed 35 out of 35 changed files in this pull request and generated 3 comments.

Comments suppressed due to low confidence (1)

janusgraph-cql/src/test/java/org/janusgraph/diskstorage/cql/CQLCdcTableOptionTest.java:1

This assertion is fairly broad and could pass if an unrelated substring containing 'cdc' appears in the DDL. Tightening it to assert the specific table option (e.g., matching cdc = true / WITH cdc = true) would make the test more robust and less prone to false positives.

Copilot

Pull request overview

Copilot reviewed 35 out of 35 changed files in this pull request and generated 4 comments.

Copilot

Pull request overview

Copilot reviewed 35 out of 35 changed files in this pull request and generated 2 comments.

Copilot

Pull request overview

Copilot reviewed 35 out of 35 changed files in this pull request and generated 8 comments.

Copilot

Pull request overview

Copilot reviewed 35 out of 35 changed files in this pull request and generated 4 comments.

Copilot

Pull request overview

Copilot reviewed 38 out of 38 changed files in this pull request and generated 1 comment.

Copilot

Pull request overview

Copilot reviewed 38 out of 38 changed files in this pull request and generated 1 comment.

Copilot

Pull request overview

Copilot reviewed 39 out of 39 changed files in this pull request and generated 3 comments.

Keep mixed indexes (ElasticSearch/Solr/Lucene) eventually consistent with the graph by deriving their updates from a Change-Data-Capture stream of the committed graph data, instead of a synchronous second write during the transaction that can diverge on failure and leave a permanently stale index. Pipeline (Apache Cassandra): commit (graph data only) -> Cassandra edgestore(cdc=true) -> Debezium -> Kafka -> CdcIndexUpdateWorker consumer group -> reindex-from-current-state -> ES bulk. Key design points: - Reindex-from-current-state: the worker reads each changed element's current graph state and fully replaces its index document (reusing IndexSerializer, like transaction recovery). This is idempotent and order-independent, so out-of-order or duplicate events still converge to the current state -- no strict ordering required, and a stale event can never overwrite a fresh value. - No dual write: the only synchronous write is to storage; the index is updated downstream from the committed change stream, so it cannot diverge. - Element-keyed Kafka partitioning gives per-element ordering and horizontal scaling via a consumer group; batches are de-duplicated and applied as one ElasticSearch _bulk per index. - At-least-once: offsets are committed only after a batch is durably applied; on failure the batch is reprocessed (rewind) rather than skipped, so the index eventually catches up. Configuration (opt-in, disabled by default): - storage.cql.cdc: emit the Cassandra cdc=true table option on the edgestore table. - index.[X].cdc.enabled / index.[X].cdc.synchronous: per-index dual mode (write synchronously AND via CDC) or cdc-only mode (skip the synchronous write; ES updated solely via CDC). Components: - janusgraph-core: per-index CDC config options, the commit-side skip hook in StandardJanusGraph (with a startup warning when cdc-only mode is configured), and MixedIndexUpdateApplier (the backend-agnostic reindex-from-current-state engine, covering vertex, edge and property-element mixed indexes). The restore paths (ElementCategory.retrieve, IndexSerializer.removeElement) now accept custom String vertex ids in addition to Long, and RelationIdentifierUtils.findRelation no longer NPEs when a relation's adjacent vertex has been removed. - janusgraph-cql: the storage.cql.cdc table option (no Kafka dependency in production code). - janusgraph-cdc (new module; core + kafka-clients): the CdcEventDecoder SPI, DebeziumCassandraJsonDecoder (parses the relation header directly, so IN-direction edge columns and value-less delete tombstones of MULTI edges resolve to the correct edge identity), the CdcWorkerConfiguration, the CdcIndexUpdateWorker (two-phase shutdown, interrupt-aware retries, no consumer leaks), and the standalone CdcIndexUpdateWorkerMain runner. The deletion limits of cdc-only mode for constrained-multiplicity edge indexes and meta-property indexes are documented. Testing: 54 tests, including unit/component coverage (decoder vs real serialized bytes incl. poison-pill skips for invalid-Base64/malformed-key, delete-envelope after=null/before fallback, IN-direction columns and value-less edge-delete tombstones; reindex engine over vertex/edge/property-element indexes incl. document removal when an element loses all indexed fields, removed-endpoint edges, and custom String vertex ids; worker loop via Kafka MockConsumer incl. at-least-once rewind on decode/apply failure and leak-free lifecycle; fail-fast config and CDC-enablement validation; full-chain convergence over Lucene incl. vertex/edge add/update/remove and out-of-order delivery) and two real-container E2Es -- worker -> Kafka -> ElasticSearch, and the full Cassandra-CDC -> Debezium -> Kafka -> ElasticSearch pipeline covering the vertex AND edge lifecycle (add/update/property-removal/delete) against real Debezium delete envelopes. The real-container tests are gated behind the cassandra-cdc-e2e Maven profile (auto-activated on Java 17+, required by Debezium 3.x and Testcontainers 2.x); the default Java 8/11 build excludes them and stays green. CI: a dedicated workflow (.github/workflows/ci-cdc.yml) runs the cdc unit tests on Java 8 and 11 and the full real-container suite -- including the Cassandra-CDC -> Debezium -> Kafka -> ElasticSearch pipeline -- on Java 17 with Docker, so the integration is exercised on every change and guards against regression. Docs: advanced-topics/cdc-mixed-index.md operator guide, a 1.2.0 changelog upgrade note, and the regenerated configuration reference. Fixes JanusGraph#4873 Replaces JanusGraph#4874 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Signed-off-by: Oleksandr Porunov <alexandr.porunov@gmail.com>

…4873) Follow-up to d81f2b7 applying the findings of a max-effort review pass. Every change, its rationale, and its verification are described in CDC-REVIEW-CHANGES.md (root). Correctness: - MixedIndexUpdateApplier opens its transactions with skipDBCacheRead(): reindex-from-current-state must read the live graph, and the worker JVM's database-level cache is never invalidated by remote writers -- a cached read could permanently miss a change (or resurrect a deleted element). - cdc.consumer.auto.offset.reset is honored as an override; "earliest" is now only the default (putIfAbsent), per the documented pass-through contract. - cdc.poll-timeout-ms must be positive: poll(ZERO) busy-spun the loop. - RelationIdentifierUtils.findEdgeRelations returns an empty Iterable instead of null (contract fixed at the source; the caller-side null guard is gone). Design: - index.[X].cdc.enabled / index.[X].cdc.synchronous are GLOBAL_OFFLINE (were MASKABLE): the commit-side skip and the worker's index discovery are one cluster-wide contract, and per-instance local values could split it (servers skipping writes while the worker manages nothing). Docs explain the mgmt.set(...) flow for existing graphs. - cdc-only mixed-index updates are filtered out at generation time via an IndexAppliesToFunction passed to getIndexUpdates, instead of being materialized (serialized) and then discarded in the write loop. has2iMods/WAL/lock semantics are unchanged; composite indexes are never filtered (new regression test). - index.[X].cdc.synchronous=false without cdc.enabled=true now logs a warning instead of being silently inert. Efficiency: - The applier applies a batch in ONE transaction spanning all backing indexes (was one per backing) with a shared element cache, and batch-preloads VERTEX changes via getVertices(...) + multiQuery() .properties() (one existence multiget + one sliced property prefetch instead of per-element round-trips); missing ids are null-cached so deletions skip the per-id read. Still one restore (one ES _bulk) per backing index. Operations & docs: - CdcIndexUpdateWorkerMain.close() logs what happens when a worker is still mid-apply after the bounded join (batch fails without committing offsets and is redelivered) instead of silently closing the graph under it. - Documented the long-outage interaction between the retry budget and Kafka's max.poll.interval.ms (group eviction churn) with tuning guidance, and the auto.offset.reset default. - Regenerated the configuration reference; updated the changelog note. Tests (Java 11 suites all green; real-container E2Es pass on Java 17+): - Edge-decoder assertions now compare full RelationIdentifier identity (endpoint ids included) -- RelationIdentifier.equals ignores endpoints, so endpoint-swap regressions were previously invisible. - The partitioned-vertex test now pins BOTH id contracts with a real partitioned hub: VERTEX changes canonicalize the partition-copy row id, while RelationIdentifier endpoints keep the raw partition-representative id (three independent review passes flagged the latter as a bug; implementing "the fix" broke this test, empirically proving raw ids are correct -- the test now prevents that regression). - New coverage: multi-backing apply, unknown-id-in-batch, composite index under cdc-only mode, auto.offset.reset default/override, zero/negative duration validation, isAlive() lifecycle; worker-main test setup deduplicated and its missing try/finally added. - ci-cdc.yml: removed the dead ES_JAVA_OPTS env (the ES test container hard-codes its own) and the unused matrix.args interpolation. Verified: janusgraph-cdc 38/38, janusgraph-lucene 15/15, janusgraph-cql 3/3, janusgraph-core CDC tests 2/2 (Java 11); CdcKafkaElasticsearchTest 2/2 and CdcCassandraDebeziumElasticsearchTest 2/2 against real containers (Java 17+ profile). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Oleksandr Porunov <alexandr.porunov@gmail.com>

porunov added kind/enhancement storage/cassandra area/core labels Jun 28, 2026

porunov requested a review from Copilot June 28, 2026 10:02

Copilot started reviewing on behalf of porunov June 28, 2026 10:02 View session

Copilot AI reviewed Jun 28, 2026

View reviewed changes

porunov mentioned this pull request Jun 28, 2026

Add CDC support for mixed index mutations via Kafka #4874

Closed

porunov force-pushed the feature/4873-cdc-mixed-index branch 4 times, most recently from a7d91de to 5659caf Compare June 30, 2026 19:51

porunov requested a review from Copilot June 30, 2026 22:25

Copilot started reviewing on behalf of porunov June 30, 2026 22:26 View session

Copilot AI reviewed Jun 30, 2026

View reviewed changes

Comment thread janusgraph-cdc/src/main/java/org/janusgraph/cdc/CdcIndexUpdateWorker.java Outdated

Comment thread janusgraph-cdc/src/main/java/org/janusgraph/cdc/DebeziumCassandraJsonDecoder.java

Comment thread janusgraph-cdc/src/main/java/org/janusgraph/cdc/CdcIndexUpdateWorkerMain.java Outdated

porunov force-pushed the feature/4873-cdc-mixed-index branch 2 times, most recently from 48c32ba to 3e6cc0e Compare July 1, 2026 10:08

porunov requested a review from Copilot July 1, 2026 10:09