You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org> on 2025/09/29 19:39:24 UTC
[PR] KAFKA-19744: Added a wrapper for InternalProcessorContext to calculate restore metrics. [kafka]
Nikita-Shupletsov opened a new pull request, #20613:
URL: https://github.com/apache/kafka/pull/20613
* Added a wrapper for InternalProcessorContext to calculate restore metrics.
* Moved the restore metric calculatiion to the restore method instead of init.
--
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-19744: Move restore time calculation to ChangelogMetadata [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #20613:
URL: https://github.com/apache/kafka/pull/20613#issuecomment-3359341503
One more thought: Can we extend `RestoreIntegrationTest`, using `TrackingStateRestoreListener` to also establish a bound on restore latency, and compare our bound to the collected metric, ie, `bound - 1 sec < metric-value< bound + 1 sec` (to allow for some jitter).
--
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-19744: Move restore time calculation to ChangelogMetadata [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #20613:
URL: https://github.com/apache/kafka/pull/20613#discussion_r2400354212
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java:
##########
@@ -1364,6 +1366,29 @@ public void shouldNotThrowOnUnknownRevokedPartition() {
}
}
+ @Test
+ public void shouldCallRecordRestoreTimeAtTheEndOfRestore() {
+ setupActiveStateManager();
+
+ final MeteredStateStore meteredStateStore = mock(MeteredStateStore.class);
+
+ when(storeMetadata.changelogPartition()).thenReturn(tp);
+ when(storeMetadata.store()).thenReturn(meteredStateStore);
+ when(meteredStateStore.name()).thenReturn(storeName);
+ final TaskId taskId = new TaskId(0, 0);
+
+ when(storeMetadata.offset()).thenReturn(5L);
+ when(activeStateManager.taskId()).thenReturn(taskId);
+
+ setupConsumer(1, tp);
+
+ changelogReader.register(tp, activeStateManager);
+
+ changelogReader.restore(Collections.singletonMap(taskId, mock(Task.class)));
Review Comment:
I updated the test. hopefully, it's a bit clearer what it's trying to test
--
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-19744: Move restore time calculation to ChangelogMetadata [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #20613:
URL: https://github.com/apache/kafka/pull/20613
--
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-19744: Move restore time calculation to ChangelogMetadata. [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #20613:
URL: https://github.com/apache/kafka/pull/20613#discussion_r2395985665
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredInternalProcessorContext.java:
##########
@@ -0,0 +1,312 @@
+/*
+ * 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.state.internals;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.streams.processor.Cancellable;
+import org.apache.kafka.streams.processor.CommitCallback;
+import org.apache.kafka.streams.processor.PunctuationType;
+import org.apache.kafka.streams.processor.Punctuator;
+import org.apache.kafka.streams.processor.StateRestoreCallback;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.To;
+import org.apache.kafka.streams.processor.api.FixedKeyRecord;
+import org.apache.kafka.streams.processor.api.Record;
+import org.apache.kafka.streams.processor.api.RecordMetadata;
+import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
+import org.apache.kafka.streams.processor.internals.ProcessorMetadata;
+import org.apache.kafka.streams.processor.internals.ProcessorNode;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
+import org.apache.kafka.streams.processor.internals.RecordCollector;
+import org.apache.kafka.streams.processor.internals.StreamTask;
+import org.apache.kafka.streams.processor.internals.Task;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.query.Position;
+import org.apache.kafka.streams.state.StoreBuilder;
+import org.apache.kafka.streams.state.internals.metrics.StateStoreMetrics;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.Map;
+import java.util.Optional;
+
+public class MeteredInternalProcessorContext<KOut, VOut> implements InternalProcessorContext<KOut, VOut> {
+
+ private final InternalProcessorContext<KOut, VOut> wrapped;
+ private final Time time;
+ private final Sensor restoreSensor;
+
+ public MeteredInternalProcessorContext(final InternalProcessorContext<KOut, VOut> wrapped, final String storeType,
+ final String storeName, final Time time) {
+ this.wrapped = wrapped;
+ this.time = time;
+ this.restoreSensor = StateStoreMetrics.restoreSensor(wrapped.taskId().toString(), storeType, storeName, wrapped.metrics());
+ }
+
+ @Override
+ public Optional<RecordMetadata> recordMetadata() {
+ return wrapped.recordMetadata();
+ }
+
+ @Override
+ public void register(final StateStore store, final StateRestoreCallback stateRestoreCallback, final CommitCallback commitCallback) {
+ wrapped.register(store, new MeteredStateRestoreCallback(stateRestoreCallback), commitCallback);
+ }
+
+ @Override
+ public <K extends KOut, V extends VOut> void forward(final Record<K, V> record) {
+ wrapped.forward(record);
+ }
+
+ @Override
+ public <K extends KOut, V extends VOut> void forward(final Record<K, V> record, final String childName) {
+ wrapped.forward(record, childName);
+ }
+
+ @Override
+ public <K extends KOut, V extends VOut> void forward(final FixedKeyRecord<K, V> record) {
+ wrapped.forward(record);
+ }
+
+ @Override
+ public <K extends KOut, V extends VOut> void forward(final FixedKeyRecord<K, V> record, final String childName) {
+ wrapped.forward(record, childName);
+ }
+
+ @Override
+ public String applicationId() {
+ return wrapped.applicationId();
+ }
+
+ @Override
+ public TaskId taskId() {
+ return wrapped.taskId();
+ }
+
+ @Override
+ public Serde<?> keySerde() {
+ return wrapped.keySerde();
+ }
+
+ @Override
+ public Serde<?> valueSerde() {
+ return wrapped.valueSerde();
+ }
+
+ @Override
+ public File stateDir() {
+ return wrapped.stateDir();
+ }
+
+ @Override
+ public void register(final StateStore store, final StateRestoreCallback stateRestoreCallback) {
+ wrapped.register(store, new MeteredStateRestoreCallback(stateRestoreCallback));
+ }
+
+ @Override
+ public <S extends StateStore> S getStateStore(final String name) {
+ return wrapped.getStateStore(name);
+ }
+
+ @Override
+ public Cancellable schedule(final Duration interval, final PunctuationType type, final Punctuator callback) {
+ return wrapped.schedule(interval, type, callback);
+ }
+
+ @Override
+ public <K, V> void forward(final K key, final V value) {
+ wrapped.forward(key, value);
+ }
+
+ @Override
+ public <K, V> void forward(final K key, final V value, final To to) {
+ wrapped.forward(key, value, to);
+ }
+
+ @Override
+ public void commit() {
+ wrapped.commit();
+ }
+
+ @Override
+ public String topic() {
+ return wrapped.topic();
+ }
+
+ @Override
+ public int partition() {
+ return wrapped.partition();
+ }
+
+ @Override
+ public long offset() {
+ return wrapped.offset();
+ }
+
+ @Override
+ public Headers headers() {
+ return wrapped.headers();
+ }
+
+ @Override
+ public long timestamp() {
+ return wrapped.timestamp();
+ }
+
+ @Override
+ public Map<String, Object> appConfigs() {
+ return wrapped.appConfigs();
+ }
+
+ @Override
+ public Map<String, Object> appConfigsWithPrefix(final String prefix) {
+ return wrapped.appConfigsWithPrefix(prefix);
+ }
+
+ @Override
+ public long currentSystemTimeMs() {
+ return wrapped.currentSystemTimeMs();
+ }
+
+ @Override
+ public long currentStreamTimeMs() {
+ return wrapped.currentStreamTimeMs();
+ }
+
+ @Override
+ public StreamsMetricsImpl metrics() {
+ return wrapped.metrics();
+ }
+
+ @Override
+ public void setSystemTimeMs(final long timeMs) {
+ wrapped.setSystemTimeMs(timeMs);
+ }
+
+ @Override
+ public ProcessorRecordContext recordContext() {
+ return wrapped.recordContext();
+ }
+
+ @Override
+ public void setRecordContext(final ProcessorRecordContext recordContext) {
+ wrapped.setRecordContext(recordContext);
+ }
+
+ @Override
+ public void setCurrentNode(final ProcessorNode<?, ?, ?, ?> currentNode) {
+ wrapped.setCurrentNode(currentNode);
+ }
+
+ @Override
+ public ProcessorNode<?, ?, ?, ?> currentNode() {
+ return wrapped.currentNode();
+ }
+
+ @Override
+ public ThreadCache cache() {
+ return wrapped.cache();
+ }
+
+ @Override
+ public void initialize() {
+ wrapped.initialize();
+ }
+
+ @Override
+ public void uninitialize() {
+ wrapped.uninitialize();
+ }
+
+ @Override
+ public Task.TaskType taskType() {
+ return wrapped.taskType();
+ }
+
+ @Override
+ public void transitionToActive(final StreamTask streamTask, final RecordCollector recordCollector, final ThreadCache newCache) {
+ wrapped.transitionToActive(streamTask, recordCollector, newCache);
+ }
+
+ @Override
+ public void transitionToStandby(final ThreadCache newCache) {
+ wrapped.transitionToStandby(newCache);
+ }
+
+ @Override
+ public void registerCacheFlushListener(final String namespace, final ThreadCache.DirtyEntryFlushListener listener) {
+ wrapped.registerCacheFlushListener(namespace, listener);
+ }
+
+ @Override
+ public <T extends StateStore> T getStateStore(final StoreBuilder<T> builder) {
+ return wrapped.getStateStore(builder);
+ }
+
+ @Override
+ public void logChange(final String storeName, final Bytes key, final byte[] value, final long timestamp, final Position position) {
+ wrapped.logChange(storeName, key, value, timestamp, position);
+ }
+
+ @Override
+ public String changelogFor(final String storeName) {
+ return wrapped.changelogFor(storeName);
+ }
+
+ @Override
+ public void addProcessorMetadataKeyValue(final String key, final long value) {
+ wrapped.addProcessorMetadataKeyValue(key, value);
+ }
+
+ @Override
+ public Long processorMetadataForKey(final String key) {
+ return wrapped.processorMetadataForKey(key);
+ }
+
+ @Override
+ public void setProcessorMetadata(final ProcessorMetadata metadata) {
+ wrapped.setProcessorMetadata(metadata);
+ }
+
+ @Override
+ public ProcessorMetadata processorMetadata() {
+ return wrapped.processorMetadata();
+ }
+
+ public InternalProcessorContext<KOut, VOut> wrapped() {
+ return wrapped;
+ }
+
+ private class MeteredStateRestoreCallback implements StateRestoreCallback {
+
+ private final StateRestoreCallback wrapped;
+
+ MeteredStateRestoreCallback(final StateRestoreCallback wrapped) {
+ this.wrapped = wrapped;
+ }
+
+ @Override
+ public void restore(final byte[] key, final byte[] value) {
+ StreamsMetricsImpl.maybeMeasureLatency(() -> wrapped.restore(key, value), time, restoreSensor);
Review Comment:
completely reworked the PR. please let me know what you think.
--
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-19744: Move restore time calculation to ChangelogMetadata. [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on PR #20613:
URL: https://github.com/apache/kafka/pull/20613#issuecomment-3358284667
> This PR updates `MeteredKeyValueStore` and `MeteredSessionStore`, but there is also `MeteredWindowStore`.
>
> There is also `MeteredVersionStore` which does not use the sensor -- so I assume, we incorrectly don't record the restore metric for versioned stores? -- We should also close this gap.
thanks for catching that.
I took a look. MeteredTimestampedKeyValueStore, MeteredTimestampedKeyValueStore, and MeteredVersionedKeyValueStoreInternal extend MeteredKeyValueStore. and MeteredTimestampedWindowStore extends MeteredWindowStore
so the only missed metered store was MeteredWindowStore. at least I wasn't able to find anything else
--
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-19744: Move restore time calculation to ChangelogMetadata. [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on PR #20613:
URL: https://github.com/apache/kafka/pull/20613#issuecomment-3359024518
> Seems a test is breaking: `FAILED ❌ MeteredVersionedKeyValueStoreTest > shouldRecordMetricsOnInit()`
>
> But given the name of the test, it might actually be void, and we should remove the entire test, as we actually don't want to record any metrics in `init()`?
yes, you are right. sorry, idk how I missed that test. deleted
--
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-19744: Move restore time calculation to ChangelogMetadata. [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20613:
URL: https://github.com/apache/kafka/pull/20613#discussion_r2396744069
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -188,6 +191,10 @@ List<ConsumerRecord<byte[], byte[]>> bufferedRecords() {
int bufferedLimitIndex() {
return bufferedLimitIndex;
}
+
+ long calculateRestoreTime(final Time time) {
Review Comment:
Why do we pass `Time` instead of `long restoreCompletionTimeNs` ? (Basic API design suggest, to pass the minimum what you need.)
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java:
##########
@@ -1364,6 +1366,29 @@ public void shouldNotThrowOnUnknownRevokedPartition() {
}
}
+ @Test
+ public void shouldCallRecordRestoreTimeAtTheEndOfRestore() {
+ setupActiveStateManager();
+
+ final MeteredStateStore meteredStateStore = mock(MeteredStateStore.class);
+
+ when(storeMetadata.changelogPartition()).thenReturn(tp);
+ when(storeMetadata.store()).thenReturn(meteredStateStore);
+ when(meteredStateStore.name()).thenReturn(storeName);
+ final TaskId taskId = new TaskId(0, 0);
+
+ when(storeMetadata.offset()).thenReturn(5L);
+ when(activeStateManager.taskId()).thenReturn(taskId);
+
+ setupConsumer(1, tp);
+
+ changelogReader.register(tp, activeStateManager);
+
+ changelogReader.restore(Collections.singletonMap(taskId, mock(Task.class)));
Review Comment:
It's not totally clear to me, why this call ensure that restore completes, and we hit the right code path?
Should we also add a negative test, `shouldNotRecordRestoreTimeIfRestoreDoesNotComplete`?
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreTest.java:
##########
@@ -458,10 +458,13 @@ public void shouldNotFindExpiredSessionRangeInBackwardOrderFromStore() {
}
@Test
- public void shouldRecordRestoreTimeOnInit() {
+ public void shouldRecordRestoreLatencyOnRecordRestoreTime() {
setUp();
init();
+ final long restoreTimeNs = 1000;
Review Comment:
```suggestion
final long restoreTimeNs = 1000L;
```
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredStateStore.java:
##########
@@ -0,0 +1,24 @@
+/*
+ * 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.state.internals;
+
+import org.apache.kafka.streams.processor.StateStore;
+
+public interface MeteredStateStore extends StateStore {
Review Comment:
Seems unnecessary (even if not harmful) to let it `extends StateStore` ?
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredWindowStoreTest.java:
##########
@@ -209,6 +209,22 @@ public void testMetrics() {
)));
}
+ @Test
+ public void shouldRecordRestoreLatencyOnRecordRestoreTime() {
+ setUp();
+ doNothing().when(innerStoreMock).init(context, store);
+
+ store.init(context, store);
+
+ final long restoreTimeNs = 1000;
Review Comment:
```suggestion
final long restoreTimeNs = 1000L;
```
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredWindowStoreTest.java:
##########
@@ -209,6 +209,22 @@ public void testMetrics() {
)));
}
+ @Test
+ public void shouldRecordRestoreLatencyOnRecordRestoreTime() {
+ setUp();
+ doNothing().when(innerStoreMock).init(context, store);
+
+ store.init(context, store);
+
+ final long restoreTimeNs = 1000;
+ store.recordRestoreTime(restoreTimeNs);
+
+ // it suffices to verify one restore metric since all restore metrics are recorded by the same sensor
+ // and the sensor is tested elsewhere
+ final KafkaMetric metric = metric("restore-rate");
+ assertThat((Double) metric.metricValue(), greaterThan(0.0));
Review Comment:
With the old code, it was not clear what latency we would measure, to we use `greaterThan`, but know we know that we should record exactly `1000`. Should we use `is(1000L)` instead?
Seems the existing tests for the other store use `assertTrue(...)` should we also update them to use `assertThat` ?
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStoreTest.java:
##########
@@ -210,12 +210,15 @@ public void testMetrics() {
}
@Test
- public void shouldRecordRestoreLatencyOnInit() {
+ public void shouldRecordRestoreLatencyOnRecordRestoreTime() {
setUp();
doNothing().when(inner).init(context, metered);
init();
+ final long restoreTimeNs = 1000;
Review Comment:
```suggestion
final long restoreTimeNs = 1000L;
```
--
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-19744: Move restore time calculation to ChangelogMetadata [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #20613:
URL: https://github.com/apache/kafka/pull/20613#discussion_r2397283872
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredStateStore.java:
##########
@@ -0,0 +1,24 @@
+/*
+ * 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.state.internals;
+
+import org.apache.kafka.streams.processor.StateStore;
+
+public interface MeteredStateStore extends StateStore {
Review Comment:
I extended StateStore for the ease of mocking. will fix that
--
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-19744: Move restore time calculation to ChangelogMetadata. [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #20613:
URL: https://github.com/apache/kafka/pull/20613#issuecomment-3358973709
Seems a test is breaking: `FAILED ❌ MeteredVersionedKeyValueStoreTest > shouldRecordMetricsOnInit()`
But given the name of the test, it might actually be void, and we should remove the entire test, as we actually don't want to record any metrics in `init()`?
--
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-19744: Added a wrapper for InternalProcessorContext to calculate restore metrics. [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20613:
URL: https://github.com/apache/kafka/pull/20613#discussion_r2389532788
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredInternalProcessorContext.java:
##########
@@ -0,0 +1,312 @@
+/*
+ * 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.state.internals;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.streams.processor.Cancellable;
+import org.apache.kafka.streams.processor.CommitCallback;
+import org.apache.kafka.streams.processor.PunctuationType;
+import org.apache.kafka.streams.processor.Punctuator;
+import org.apache.kafka.streams.processor.StateRestoreCallback;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.To;
+import org.apache.kafka.streams.processor.api.FixedKeyRecord;
+import org.apache.kafka.streams.processor.api.Record;
+import org.apache.kafka.streams.processor.api.RecordMetadata;
+import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
+import org.apache.kafka.streams.processor.internals.ProcessorMetadata;
+import org.apache.kafka.streams.processor.internals.ProcessorNode;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
+import org.apache.kafka.streams.processor.internals.RecordCollector;
+import org.apache.kafka.streams.processor.internals.StreamTask;
+import org.apache.kafka.streams.processor.internals.Task;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.query.Position;
+import org.apache.kafka.streams.state.StoreBuilder;
+import org.apache.kafka.streams.state.internals.metrics.StateStoreMetrics;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.Map;
+import java.util.Optional;
+
+public class MeteredInternalProcessorContext<KOut, VOut> implements InternalProcessorContext<KOut, VOut> {
+
+ private final InternalProcessorContext<KOut, VOut> wrapped;
+ private final Time time;
+ private final Sensor restoreSensor;
+
+ public MeteredInternalProcessorContext(final InternalProcessorContext<KOut, VOut> wrapped, final String storeType,
+ final String storeName, final Time time) {
+ this.wrapped = wrapped;
+ this.time = time;
+ this.restoreSensor = StateStoreMetrics.restoreSensor(wrapped.taskId().toString(), storeType, storeName, wrapped.metrics());
+ }
+
+ @Override
+ public Optional<RecordMetadata> recordMetadata() {
+ return wrapped.recordMetadata();
+ }
+
+ @Override
+ public void register(final StateStore store, final StateRestoreCallback stateRestoreCallback, final CommitCallback commitCallback) {
+ wrapped.register(store, new MeteredStateRestoreCallback(stateRestoreCallback), commitCallback);
+ }
+
+ @Override
+ public <K extends KOut, V extends VOut> void forward(final Record<K, V> record) {
+ wrapped.forward(record);
+ }
+
+ @Override
+ public <K extends KOut, V extends VOut> void forward(final Record<K, V> record, final String childName) {
+ wrapped.forward(record, childName);
+ }
+
+ @Override
+ public <K extends KOut, V extends VOut> void forward(final FixedKeyRecord<K, V> record) {
+ wrapped.forward(record);
+ }
+
+ @Override
+ public <K extends KOut, V extends VOut> void forward(final FixedKeyRecord<K, V> record, final String childName) {
+ wrapped.forward(record, childName);
+ }
+
+ @Override
+ public String applicationId() {
+ return wrapped.applicationId();
+ }
+
+ @Override
+ public TaskId taskId() {
+ return wrapped.taskId();
+ }
+
+ @Override
+ public Serde<?> keySerde() {
+ return wrapped.keySerde();
+ }
+
+ @Override
+ public Serde<?> valueSerde() {
+ return wrapped.valueSerde();
+ }
+
+ @Override
+ public File stateDir() {
+ return wrapped.stateDir();
+ }
+
+ @Override
+ public void register(final StateStore store, final StateRestoreCallback stateRestoreCallback) {
+ wrapped.register(store, new MeteredStateRestoreCallback(stateRestoreCallback));
+ }
+
+ @Override
+ public <S extends StateStore> S getStateStore(final String name) {
+ return wrapped.getStateStore(name);
+ }
+
+ @Override
+ public Cancellable schedule(final Duration interval, final PunctuationType type, final Punctuator callback) {
+ return wrapped.schedule(interval, type, callback);
+ }
+
+ @Override
+ public <K, V> void forward(final K key, final V value) {
+ wrapped.forward(key, value);
+ }
+
+ @Override
+ public <K, V> void forward(final K key, final V value, final To to) {
+ wrapped.forward(key, value, to);
+ }
+
+ @Override
+ public void commit() {
+ wrapped.commit();
+ }
+
+ @Override
+ public String topic() {
+ return wrapped.topic();
+ }
+
+ @Override
+ public int partition() {
+ return wrapped.partition();
+ }
+
+ @Override
+ public long offset() {
+ return wrapped.offset();
+ }
+
+ @Override
+ public Headers headers() {
+ return wrapped.headers();
+ }
+
+ @Override
+ public long timestamp() {
+ return wrapped.timestamp();
+ }
+
+ @Override
+ public Map<String, Object> appConfigs() {
+ return wrapped.appConfigs();
+ }
+
+ @Override
+ public Map<String, Object> appConfigsWithPrefix(final String prefix) {
+ return wrapped.appConfigsWithPrefix(prefix);
+ }
+
+ @Override
+ public long currentSystemTimeMs() {
+ return wrapped.currentSystemTimeMs();
+ }
+
+ @Override
+ public long currentStreamTimeMs() {
+ return wrapped.currentStreamTimeMs();
+ }
+
+ @Override
+ public StreamsMetricsImpl metrics() {
+ return wrapped.metrics();
+ }
+
+ @Override
+ public void setSystemTimeMs(final long timeMs) {
+ wrapped.setSystemTimeMs(timeMs);
+ }
+
+ @Override
+ public ProcessorRecordContext recordContext() {
+ return wrapped.recordContext();
+ }
+
+ @Override
+ public void setRecordContext(final ProcessorRecordContext recordContext) {
+ wrapped.setRecordContext(recordContext);
+ }
+
+ @Override
+ public void setCurrentNode(final ProcessorNode<?, ?, ?, ?> currentNode) {
+ wrapped.setCurrentNode(currentNode);
+ }
+
+ @Override
+ public ProcessorNode<?, ?, ?, ?> currentNode() {
+ return wrapped.currentNode();
+ }
+
+ @Override
+ public ThreadCache cache() {
+ return wrapped.cache();
+ }
+
+ @Override
+ public void initialize() {
+ wrapped.initialize();
+ }
+
+ @Override
+ public void uninitialize() {
+ wrapped.uninitialize();
+ }
+
+ @Override
+ public Task.TaskType taskType() {
+ return wrapped.taskType();
+ }
+
+ @Override
+ public void transitionToActive(final StreamTask streamTask, final RecordCollector recordCollector, final ThreadCache newCache) {
+ wrapped.transitionToActive(streamTask, recordCollector, newCache);
+ }
+
+ @Override
+ public void transitionToStandby(final ThreadCache newCache) {
+ wrapped.transitionToStandby(newCache);
+ }
+
+ @Override
+ public void registerCacheFlushListener(final String namespace, final ThreadCache.DirtyEntryFlushListener listener) {
+ wrapped.registerCacheFlushListener(namespace, listener);
+ }
+
+ @Override
+ public <T extends StateStore> T getStateStore(final StoreBuilder<T> builder) {
+ return wrapped.getStateStore(builder);
+ }
+
+ @Override
+ public void logChange(final String storeName, final Bytes key, final byte[] value, final long timestamp, final Position position) {
+ wrapped.logChange(storeName, key, value, timestamp, position);
+ }
+
+ @Override
+ public String changelogFor(final String storeName) {
+ return wrapped.changelogFor(storeName);
+ }
+
+ @Override
+ public void addProcessorMetadataKeyValue(final String key, final long value) {
+ wrapped.addProcessorMetadataKeyValue(key, value);
+ }
+
+ @Override
+ public Long processorMetadataForKey(final String key) {
+ return wrapped.processorMetadataForKey(key);
+ }
+
+ @Override
+ public void setProcessorMetadata(final ProcessorMetadata metadata) {
+ wrapped.setProcessorMetadata(metadata);
+ }
+
+ @Override
+ public ProcessorMetadata processorMetadata() {
+ return wrapped.processorMetadata();
+ }
+
+ public InternalProcessorContext<KOut, VOut> wrapped() {
+ return wrapped;
+ }
+
+ private class MeteredStateRestoreCallback implements StateRestoreCallback {
+
+ private final StateRestoreCallback wrapped;
+
+ MeteredStateRestoreCallback(final StateRestoreCallback wrapped) {
+ this.wrapped = wrapped;
+ }
+
+ @Override
+ public void restore(final byte[] key, final byte[] value) {
+ StreamsMetricsImpl.maybeMeasureLatency(() -> wrapped.restore(key, value), time, restoreSensor);
Review Comment:
The call to `restore()` does restore one "batch" of data, ie, whatever a single `poll()` call returned. So this is not the latency from "begin restore" to "complete restore".
We could maybe use `StateRestoreListener` callback, but also not sure if that's the best way?
I don't think we could use `maybeMeasureLatency()` helper method at all, but would need to get a ts when restore stars, and also when restore completes, and call `sensor.record(...)` directly?
This would go into StoreChangelogReader? Or we use existing `Task#recordRestoration()` method (or add similar overloads). We need to break it out per-store 🤔
--
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