guozhangwang commented on a change in pull request #11676:
URL: https://github.com/apache/kafka/pull/11676#discussion_r790358840
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -111,12 +112,12 @@ public StreamsMetricsImpl metrics() {
@Override
public void register(final StateStore store,
- final StateRestoreCallback stateRestoreCallback) {
+ final StateRestoreCallback stateRestoreCallback,
final CheckpointCallback checkpoint) {
Review comment:
nit: let's use a newline for each parameter, or use a single line for
all parameters if they are short.
Also: I'd suggest we rename the parameter to `checkpointCallback`, to be
more readable.
Ditto elsewhere.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java
##########
@@ -152,4 +153,11 @@ default void init(final StateStoreContext context, final
StateStore root) {
// If a store doesn't implement a query handler, then all queries are
unknown.
return QueryResult.forUnknownQueryType(query, this);
}
+
+ /**
+ * This state store's position
+ */
+ default Position getPosition() {
+ throw new UnsupportedOperationException("Not implemented");
Review comment:
Hmm.. do we need it to be a public API? I thought that with the
callback, we do not need to expose it to public APIs (for getting the position
from wrapped stores we can do that via internal functions), i.e. we only need
to do either of them to achieve our goal?
##########
File path:
streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStore.java
##########
@@ -264,8 +267,12 @@ public void init(final ProcessorContext context,
segments.openExisting(this.context, observedStreamTime);
+ this.positionCheckpointFile = new File(context.stateDir(), this.name()
+ ".position");
Review comment:
The logic of reading checkpoint upon init is duplicated on each of the
classes, could we consolidate them further into `StoreQueryUtils`, e.g. by just
passing in the stateDIR/stateName in it?
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/CheckpointCallback.java
##########
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.streams.processor;
+
+import java.io.IOException;
+
+/**
+ * Callback for checkpointing position information in state stores.
+ */
+public interface CheckpointCallback {
Review comment:
See my other comment: how about extending this callback to sth. like
"CommitCallback"?
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java
##########
@@ -85,8 +85,23 @@
* @throws IllegalStateException If store gets registered after
initialized is already finished
* @throws StreamsException if the store's change log does not contain the
partition
*/
+ default void register(final StateStore store,
+ final StateRestoreCallback stateRestoreCallback) {
+ register(store, stateRestoreCallback, null);
+ }
+
+ /**
+ * Register and possibly restores the specified storage engine.
+ *
+ * @param store the storage engine
+ * @param stateRestoreCallback the restoration callback logic for
log-backed state stores upon restart
+ * @param checkpointCallback called to checkpoint position metatadata of
state stores
+ *
+ * @throws IllegalStateException If store gets registered after
initialized is already finished
+ * @throws StreamsException if the store's change log does not contain the
partition
+ */
void register(final StateStore store,
- final StateRestoreCallback stateRestoreCallback);
+ final StateRestoreCallback stateRestoreCallback, final
CheckpointCallback checkpointCallback);
Review comment:
This is just for the KIP discussion itself: WYDT about generalizing the
name of the callback to sth like `CommitCallback`, where the semantics is that
the callback would be triggered whenever the corresponding task is committed.
My motivation is that:
* Today the state store "checkpoint" is an internal implementation detail
which is abstracted away from users. While "commit" is a public concept that
users understand. It's triggering event is different from `StateStore#flush`
since a state store would always be flushed when committing, but not vice versa.
* In the future, we may or may not need to always persist the position
metadata upon checkpointing the state stores.
* But we may want to expose such a callback for other purposes, e.g. for
in-memory stores checkpoint its data into disks along with the checkpoint file
so we do not need to re-load from changelog, or for KS to checkpoint (not the
same "checkpoint" as we used the term today, but to really checkpoint the state
itself, sync or async) to remote storage etc.
Just at the moment, we use this callback for writing position metadata; but
keeping its name generalized allow us to modify the built-in store's logic
without changing the public APIs.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java
##########
@@ -85,8 +85,23 @@
* @throws IllegalStateException If store gets registered after
initialized is already finished
* @throws StreamsException if the store's change log does not contain the
partition
*/
+ default void register(final StateStore store,
+ final StateRestoreCallback stateRestoreCallback) {
+ register(store, stateRestoreCallback, null);
+ }
+
+ /**
+ * Register and possibly restores the specified storage engine.
+ *
+ * @param store the storage engine
+ * @param stateRestoreCallback the restoration callback logic for
log-backed state stores upon restart
+ * @param checkpointCallback called to checkpoint position metatadata of
state stores
+ *
+ * @throws IllegalStateException If store gets registered after
initialized is already finished
+ * @throws StreamsException if the store's change log does not contain the
partition
+ */
void register(final StateStore store,
- final StateRestoreCallback stateRestoreCallback);
+ final StateRestoreCallback stateRestoreCallback, final
CheckpointCallback checkpointCallback);
Review comment:
nit: could we use a new line for the third parameter?
##########
File path:
streams/src/main/java/org/apache/kafka/streams/state/internals/StoreQueryUtils.java
##########
@@ -333,6 +339,43 @@ public static boolean isPermitted(
return byteArray -> deserializer.deserialize(serdes.topic(),
byteArray);
}
+
+ public static Map<TopicPartition, Long> positionToTopicPartitionMap(final
Position position) {
+ final Map<TopicPartition, Long> topicPartitions = new HashMap<>();
+ final Set<String> topics = position.getTopics();
+ for (final String t : topics) {
+ final Map<Integer, Long> partitions =
position.getPartitionPositions(t);
+ for (final Entry<Integer, Long> e : partitions.entrySet()) {
+ final TopicPartition tp = new TopicPartition(t, e.getKey());
+ topicPartitions.put(tp, e.getValue());
+ }
+ }
+ return topicPartitions;
+ }
+
+ public static void checkpointPosition(final OffsetCheckpoint
checkpointFile, final Position position) throws IOException {
+ final Map<TopicPartition, Long> topicPartitions =
StoreQueryUtils.positionToTopicPartitionMap(position);
+ checkpointFile.write(topicPartitions);
+ }
+
+ public static Position readPositionFromCheckpoint(final OffsetCheckpoint
checkpointFile) {
+ try {
+ final Map<TopicPartition, Long> topicPartitions =
checkpointFile.read();
+ return
StoreQueryUtils.topicPartitionMapToPosition(topicPartitions);
+ } catch (final IOException e) {
Review comment:
+1
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
