This is an automated email from the ASF dual-hosted git repository.
bbejeck pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 15321410af3 KAFKA-20760: Commit position atomically with changelog
offsets (#22741)
15321410af3 is described below
commit 15321410af3f349f13cf3248372fb7a87a5ef543
Author: Nick Telford <[email protected]>
AuthorDate: Fri Jul 3 01:50:18 2026 +0100
KAFKA-20760: Commit position atomically with changelog offsets (#22741)
Under EOS, a transactional `RocksDBStore` persisted its `Position` in a
separate `WriteBatch` flush from the one that persisted the changelog
offsets. A crash between the two flushes could leave the on-disk
position inconsistent with the stored offsets, so on restart the store
could report a position that does not match the data it actually holds.
This change merges the two `commit()` overloads into a single method
that takes both the in-memory `Position` and the changelog offsets. When
offsets are present, the position (merged with any uncommitted deltas)
is written into the *same* staged `WriteBatch` as the offset entries
before `commitStagedWrites()` runs, so the data, offsets, and position
all become durable atomically. The in-memory position is advanced only
after the flush succeeds, so a failure leaves both in-memory and on-disk
state consistent with the previous commit.
Tests that previously passed an empty changelog-offsets map — which now
takes the wipe-offsets path — are updated to pass a sentinel offset so
the position is persisted as intended.
Part of a series of fixes for interactive-query position consistency
with transactional state stores.
Reviewers: Bill Bejeck <[email protected]>
Co-authored-by: Bill Bejeck <[email protected]>
---
.../state/internals/AbstractColumnFamilyAccessor.java | 19 +++++++++++++------
.../kafka/streams/state/internals/RocksDBStore.java | 18 ++++++------------
.../internals/AbstractColumnFamilyAccessorTest.java | 10 +++++-----
3 files changed, 24 insertions(+), 23 deletions(-)
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractColumnFamilyAccessor.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractColumnFamilyAccessor.java
index aebcecd75a7..bb872f58ac8 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractColumnFamilyAccessor.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractColumnFamilyAccessor.java
@@ -58,8 +58,12 @@ abstract class AbstractColumnFamilyAccessor implements
RocksDBStore.ColumnFamily
}
@Override
- public final void commit(final RocksDBStore.DBAccessor accessor, final
Map<TopicPartition, Long> changelogOffsets) throws RocksDBException {
+ public final void commit(final RocksDBStore.DBAccessor accessor,
+ final Position position,
+ final Map<TopicPartition, Long> changelogOffsets)
throws RocksDBException {
if (changelogOffsets.isEmpty()) {
+ // No changelog: wipe any stale changelog-offset keys left from a
prior
+ // logging configuration, then write the current position below.
wipeOffsets(accessor);
} else {
for (final Map.Entry<TopicPartition, Long> entry :
changelogOffsets.entrySet()) {
@@ -74,12 +78,15 @@ abstract class AbstractColumnFamilyAccessor implements
RocksDBStore.ColumnFamily
}
}
}
+ // Merge committed and uncommitted positions into a temporary copy so
that if
+ // commitStagedWrites() throws, the in-memory position remains
consistent with the
+ // last successful commit. Position is written after wipeOffsets so
the key is not
+ // deleted by the wipe. The in-memory position is updated only on
success.
+ final Position merged = position.copy();
+ accessor.mergeUncommittedPositionInto(merged);
+ accessor.put(offsetColumnFamilyHandle, positionKey,
PositionSerde.serialize(merged).array());
accessor.commitStagedWrites();
- }
-
- @Override
- public final void commit(final RocksDBStore.DBAccessor accessor, final
Position storePosition) throws RocksDBException {
- accessor.put(offsetColumnFamilyHandle, positionKey,
PositionSerde.serialize(storePosition).array());
+ position.merge(merged);
}
@Override
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
index 4edd65ae8cd..e7b2eef3a4c 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
@@ -186,7 +186,7 @@ public class RocksDBStore implements KeyValueStore<Bytes,
byte[]>, BatchWritingS
stateStoreContext.register(
root,
(RecordBatchingStateRestoreCallback) this::restoreBatch,
- this::writePosition
+ this::writePosition
);
consistencyEnabled = StreamsConfig.InternalConfig.getBoolean(
stateStoreContext.appConfigs(),
@@ -455,12 +455,7 @@ public class RocksDBStore implements KeyValueStore<Bytes,
byte[]>, BatchWritingS
}
public final void writePosition() {
- validateStoreOpen();
- try {
- cfAccessor.commit(dbAccessor, position);
- } catch (final RocksDBException e) {
- log.warn("Error while committing position for store {}", name, e);
- }
+ // Position is now committed atomically inside commit(); this method
is a no-op.
}
@Override
@@ -916,8 +911,7 @@ public class RocksDBStore implements KeyValueStore<Bytes,
byte[]>, BatchWritingS
}
try {
synchronized (position) {
- cfAccessor.commit(dbAccessor, changelogOffsets);
- dbAccessor.mergeUncommittedPositionInto(position);
+ cfAccessor.commit(dbAccessor, position, changelogOffsets);
}
} catch (final RocksDBException e) {
throw new ProcessorStateException("Error while executing commit
from store " + name, e);
@@ -1366,9 +1360,9 @@ public class RocksDBStore implements KeyValueStore<Bytes,
byte[]>, BatchWritingS
long approximateNumEntries(final DBAccessor accessor) throws
RocksDBException;
- void commit(final DBAccessor accessor, final Map<TopicPartition, Long>
changelogOffsets) throws RocksDBException;
-
- void commit(final DBAccessor accessor, final Position storePosition)
throws RocksDBException;
+ void commit(final DBAccessor accessor,
+ final Position position,
+ final Map<TopicPartition, Long> changelogOffsets) throws
RocksDBException;
void addToBatch(final byte[] key,
final byte[] value,
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractColumnFamilyAccessorTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractColumnFamilyAccessorTest.java
index d7fbf8dfa75..9473f999041 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractColumnFamilyAccessorTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractColumnFamilyAccessorTest.java
@@ -119,7 +119,7 @@ abstract class AbstractColumnFamilyAccessorTest {
final TopicPartition tp0 = new TopicPartition("testTopic", 0);
final TopicPartition tp1 = new TopicPartition("testTopic", 1);
final Map<TopicPartition, Long> changelogOffsets = Map.of(tp0, 10L,
tp1, 20L);
- accessor.commit(dbAccessor, changelogOffsets);
+ accessor.commit(dbAccessor, Position.emptyPosition(),
changelogOffsets);
assertEquals(10L, accessor.getCommittedOffset(dbAccessor, tp0));
assertEquals(20L, accessor.getCommittedOffset(dbAccessor, tp1));
}
@@ -131,14 +131,14 @@ abstract class AbstractColumnFamilyAccessorTest {
final TopicPartition tp0 = new TopicPartition(topic, 0);
final TopicPartition tp1 = new TopicPartition(topic, 1);
final Position positionToStore = Position.fromMap(mkMap(mkEntry(topic,
mkMap(mkEntry(tp0.partition(), 10L), mkEntry(tp1.partition(), 20L)))));
- accessor.commit(dbAccessor, positionToStore);
+ accessor.commit(dbAccessor, positionToStore, Map.of(tp0, 0L));
assertEquals(positionToStore,
PositionSerde.deserialize(ByteBuffer.wrap(dbAccessor.get(offsetsCF,
toBytes("position")))));
}
@Test
public void shouldCommitStagedWritesWhenCommittingOffsets() throws
RocksDBException {
final TopicPartition tp0 = new TopicPartition("testTopic", 0);
- accessor.commit(dbAccessor, Map.of(tp0, 10L));
+ accessor.commit(dbAccessor, Position.emptyPosition(), Map.of(tp0,
10L));
verify(dbAccessor).commitStagedWrites();
}
@@ -147,10 +147,10 @@ abstract class AbstractColumnFamilyAccessorTest {
dbAccessor = new InMemoryRocksDBAccessor(mock(RocksDB.class));
final TopicPartition tp0 = new TopicPartition("testTopic", 0);
final TopicPartition tp1 = new TopicPartition("testTopic", 1);
- accessor.commit(dbAccessor, Map.of(tp0, 10L, tp1, 20L));
+ accessor.commit(dbAccessor, Position.emptyPosition(), Map.of(tp0, 10L,
tp1, 20L));
assertEquals(10L, accessor.getCommittedOffset(dbAccessor, tp0));
assertEquals(20L, accessor.getCommittedOffset(dbAccessor, tp1));
- accessor.commit(dbAccessor, Map.of());
+ accessor.commit(dbAccessor, Position.emptyPosition(), Map.of());
assertNull(accessor.getCommittedOffset(dbAccessor, tp0));
assertNull(accessor.getCommittedOffset(dbAccessor, tp1));
}
