You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "nicktelford (via GitHub)" <gi...@apache.org> on 2025/11/21 18:23:58 UTC
[PR] KAFKA-14711: Add new StateStore interfaces [kafka]
nicktelford opened a new pull request, #20955:
URL: https://github.com/apache/kafka/pull/20955
These are the new interfaces detailed in KIP-1035: "StateStore managed changelog offsets".
This PR introduces the interfaces changes, but makes otherwise no consequential behavioural changes.
Outside of `StateStore.java`, _all_ changes are essentially just a rename of all invocations of `StateStore#flush` to instead call `StateStore#commit`.
The `changelogOffsets` being passed in these invocations is currently unused: the behaviour of `StateStore#commit` remains identical to `StateStore#flush` before these changes.
A new implementation of `StateStore#commit` that actually uses these offsets, along with changes to the use-site (in `ProcessorStateManager` and `GlobalStateManager`) will come in a later PR.
Many strings, including documentation, and some variable names, have also been renamed (from "flush" to "commit"), to maintain consistency with the method they relate to.
One exception is the `flush-rate` metric, which has not been renamed, because it will instead be deprecated in favour of a new `commit-rate` metric, which will be introduced in another PR.
---
The only change in behaviour is as follows: calling `StateStore#flush` from within a `Processor` is now a guaranteed no-op.
In the future, this will throw an `UnsupportedOperationException`, but to ensure no changes to end-user experience, we currently make it a no-op.
Previously, any call to `StateStore#flush` from a `Processor` would have made no difference to program semantics, but likely would introduce performance problems for RocksDB. This is because it would force a flush of RocksDB memtables to disk on every invocation, which if naively used could be on _every_ `Record`.
Consequently, making this a no-op should not make a difference for end-users, except potentially improving performance if they were incorrectly calling this method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17411: Add new StateStore interfaces [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on code in PR #20955:
URL: https://github.com/apache/kafka/pull/20955#discussion_r2765086651
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractReadWriteDecorator.java:
##########
@@ -48,6 +50,11 @@ public void init(final StateStoreContext stateStoreContext,
throw new UnsupportedOperationException(ERROR_MESSAGE);
}
+ @Override
+ public void commit(final Map<TopicPartition, Long> changelogOffsets) {
Review Comment:
I see what's happened here. At some point, nearly all the integration tests were moved out into a separate `:streams:integration-tests` gradle module, and I didn't notice. I kept running `:streams:integrationTest` blissfully unaware that I wasn't actually running them :laughing:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14711: Add new StateStore interfaces [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20955:
URL: https://github.com/apache/kafka/pull/20955#issuecomment-3590948686
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17411: Add new StateStore interfaces [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on PR #20955:
URL: https://github.com/apache/kafka/pull/20955#issuecomment-3847950694
@bbejeck @eduwercamacaro Thanks for the reviews. Rebased as requested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14711: Add new StateStore interfaces [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20955:
URL: https://github.com/apache/kafka/pull/20955#issuecomment-3624484820
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17411: Add new StateStore interfaces [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck merged PR #20955:
URL: https://github.com/apache/kafka/pull/20955
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17411: Add new StateStore interfaces [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20955:
URL: https://github.com/apache/kafka/pull/20955#discussion_r2789772079
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java:
##########
@@ -951,7 +952,8 @@ public long approximateNumEntries(final DBAccessor accessor) throws RocksDBExcep
}
@Override
- public void flush(final DBAccessor accessor) throws RocksDBException {
+ public void commit(final DBAccessor accessor,
+ final Map<TopicPartition, Long> changelogOffsets) throws RocksDBException {
accessor.flush(columnFamily);
Review Comment:
Wondering why this `flush(columnFamily)` call is not updated to `commit(changelogOffsets)` or similar?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17411: Add new StateStore interfaces [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #20955:
URL: https://github.com/apache/kafka/pull/20955#discussion_r2771461842
##########
streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java:
##########
@@ -73,8 +76,55 @@ public interface StateStore {
/**
* Flush any cached data
+ *
+ * @deprecated Use {@link org.apache.kafka.streams.processor.api.ProcessingContext#commit() ProcessorContext#commit()}
+ * instead.
+ */
+ @Deprecated
+ default void flush() {
+ // no-op
+ }
+
+ /**
+ * Commit all written records to this StateStore.
+ * <p>
+ * This method <b>CANNOT<b> be called by users from {@link org.apache.kafka.streams.processor.api.Processor
Review Comment:
```suggestion
* This method <b>CANNOT</b> be called by users from {@link org.apache.kafka.streams.processor.api.Processor
```
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java:
##########
@@ -484,36 +484,36 @@ void restore(final StateStoreMetadata storeMetadata, final List<ConsumerRecord<b
/**
* @throws TaskMigratedException recoverable error sending changelog records that would cause the task to be removed
- * @throws StreamsException fatal error when flushing the state store, for example sending changelog records failed
- * or flushing state store get IO errors; such error should cause the thread to die
+ * @throws StreamsException fatal error when commmitting the state store, for example sending changelog records failed
Review Comment:
```suggestion
* @throws StreamsException fatal error when committing the state store, for example sending changelog records failed
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17411: Add new StateStore interfaces [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20955:
URL: https://github.com/apache/kafka/pull/20955#discussion_r2801776653
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStoreTest.java:
##########
@@ -224,10 +224,10 @@ public void shouldDelegateAndRecordMetricsOnGetWithTimestamp() {
}
@Test
- public void shouldDelegateAndRecordMetricsOnFlush() {
- store.flush();
+ public void shouldDelegateAndRecordMetricsOnCommit() {
+ store.commit(Map.of());
- verify(inner).flush();
+ verify(inner).commit(Map.of());
assertThat((Double) getMetric("flush-rate").metricValue(), greaterThan(0.0));
Review Comment:
@nicktelford @bbejeck -- just came across this code on some other PR review, and wondering if we need to update this, and also check if the new `commit-rate` metric was bumped?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17411: Add new StateStore interfaces [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20955:
URL: https://github.com/apache/kafka/pull/20955#discussion_r2801776653
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStoreTest.java:
##########
@@ -224,10 +224,10 @@ public void shouldDelegateAndRecordMetricsOnGetWithTimestamp() {
}
@Test
- public void shouldDelegateAndRecordMetricsOnFlush() {
- store.flush();
+ public void shouldDelegateAndRecordMetricsOnCommit() {
+ store.commit(Map.of());
- verify(inner).flush();
+ verify(inner).commit(Map.of());
assertThat((Double) getMetric("flush-rate").metricValue(), greaterThan(0.0));
Review Comment:
@nicktelford @bbejeck -- just came across this code on some other PR review, and wondering if we need to update this, and also check if the new `commit-rate` metric was bumped?
Same question for other store types?
We are just adding a new metered store and corresponding test -- if this will be changed in a follow up PR, please make sure to also cover the newly added `MeteredTimestampedKeyValueStoreWithHeadersTest`
Cf https://github.com/apache/kafka/pull/21451
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17411: Add new StateStore interfaces [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20955:
URL: https://github.com/apache/kafka/pull/20955#discussion_r2801776653
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStoreTest.java:
##########
@@ -224,10 +224,10 @@ public void shouldDelegateAndRecordMetricsOnGetWithTimestamp() {
}
@Test
- public void shouldDelegateAndRecordMetricsOnFlush() {
- store.flush();
+ public void shouldDelegateAndRecordMetricsOnCommit() {
+ store.commit(Map.of());
- verify(inner).flush();
+ verify(inner).commit(Map.of());
assertThat((Double) getMetric("flush-rate").metricValue(), greaterThan(0.0));
Review Comment:
@nicktelford @bbejeck -- just came across this code on some other PR review, and wondering if we need to update this, and also check if the new `commit-rate` metric was bumped?
Same question for other store types?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17411: Add new StateStore interfaces [kafka]
Posted by "eduwercamacaro (via GitHub)" <gi...@apache.org>.
eduwercamacaro commented on code in PR #20955:
URL: https://github.com/apache/kafka/pull/20955#discussion_r2764534194
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractReadWriteDecorator.java:
##########
@@ -48,6 +50,11 @@ public void init(final StateStoreContext stateStoreContext,
throw new UnsupportedOperationException(ERROR_MESSAGE);
}
+ @Override
+ public void commit(final Map<TopicPartition, Long> changelogOffsets) {
Review Comment:
@nicktelford I just noticed that this integration test is broken: `org.apache.kafka.streams.integration.StandbyTaskEOSIntegrationTest.shouldWipeOutStandbyStateDirectoryIfCheckpointIsMissing`
I think you need to override the `commit()` method on the `KeyValueStoreReadWriteDecorator` inner class and delegate the call to the wrapped store. That class is defined here on the line 81.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17411: Add new StateStore interfaces [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on code in PR #20955:
URL: https://github.com/apache/kafka/pull/20955#discussion_r2764849405
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractReadWriteDecorator.java:
##########
@@ -48,6 +50,11 @@ public void init(final StateStoreContext stateStoreContext,
throw new UnsupportedOperationException(ERROR_MESSAGE);
}
+ @Override
+ public void commit(final Map<TopicPartition, Long> changelogOffsets) {
Review Comment:
Hmm, not sure how I missed this. I must have not run the full test suite properly!
The correct solution here is to remove the call to `store.commit(Map.of())` from the `Processor` used in the integration test, as `commit` should never be called from a user's `Processor`.
There looks to be a similar bug in `EosIntegrationTest` too. I'll fix them both and re-run the full integration test suite.
Thanks for spotting this!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17411: Add new StateStore interfaces [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #20955:
URL: https://github.com/apache/kafka/pull/20955#issuecomment-3861579909
merged #20955 into trunk
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org