You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2021/02/11 00:47:35 UTC
[kafka] branch 2.7 updated: KAFKA-12272: Fix commit-interval
metrics (#10104)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch 2.7
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/2.7 by this push:
new 75bf77e KAFKA-12272: Fix commit-interval metrics (#10104)
75bf77e is described below
commit 75bf77e50ee612968963ea073cec0137ef0564a8
Author: Matthias J. Sax <ma...@confluent.io>
AuthorDate: Wed Feb 10 16:44:43 2021 -0800
KAFKA-12272: Fix commit-interval metrics (#10104)
Reviewer: A. Sophie Blee-Goldman <so...@confluent.io>
---
.../streams/processor/internals/StreamThread.java | 3 +-
.../processor/internals/StreamThreadTest.java | 121 +++++++++++++++++++--
2 files changed, 113 insertions(+), 11 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
index 863289d..12f1e26 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
@@ -687,9 +687,10 @@ public class StreamThread extends Thread {
log.debug("{} punctuators ran.", punctuated);
+ final long beforeCommitMs = now;
final int committed = maybeCommit();
totalCommittedSinceLastSummary += committed;
- final long commitLatency = advanceNowAndComputeLatency();
+ final long commitLatency = Math.max(now - beforeCommitMs, 0);
totalCommitLatency += commitLatency;
if (committed > 0) {
commitSensor.record(commitLatency / (double) committed, now);
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
index e1b4b11..be2b567 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
@@ -61,9 +61,7 @@ import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.TaskMetadata;
import org.apache.kafka.streams.processor.ThreadMetadata;
-import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.ProcessorSupplier;
-import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender;
import org.apache.kafka.streams.state.KeyValueStore;
@@ -115,6 +113,7 @@ import static org.easymock.EasyMock.anyObject;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.expectLastCall;
import static org.easymock.EasyMock.mock;
+import static org.easymock.EasyMock.niceMock;
import static org.easymock.EasyMock.verify;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.not;
@@ -788,6 +787,111 @@ public class StreamThreadTest {
}
@Test
+ public void shouldRecordCommitLatency() {
+ final Consumer<byte[], byte[]> consumer = EasyMock.createNiceMock(Consumer.class);
+ expect(consumer.poll(anyObject())).andStubReturn(new ConsumerRecords<>(Collections.emptyMap()));
+ final Task task = niceMock(Task.class);
+ expect(task.id()).andStubReturn(task1);
+ expect(task.inputPartitions()).andStubReturn(Collections.singleton(t1p1));
+ final ActiveTaskCreator activeTaskCreator = mock(ActiveTaskCreator.class);
+ expect(activeTaskCreator.createTasks(anyObject(), anyObject())).andStubReturn(Collections.singleton(task));
+ expect(activeTaskCreator.producerClientIds()).andStubReturn(Collections.singleton("producerClientId"));
+ EasyMock.replay(consumer, task, activeTaskCreator);
+
+ final StreamsMetricsImpl streamsMetrics =
+ new StreamsMetricsImpl(metrics, CLIENT_ID, StreamsConfig.METRICS_LATEST, mockTime);
+
+ final TaskManager taskManager = new TaskManager(
+ null,
+ null,
+ null,
+ activeTaskCreator,
+ null,
+ internalTopologyBuilder,
+ null,
+ null,
+ null
+ ) {
+ @Override
+ int commit(final Collection<Task> tasksToCommit) {
+ mockTime.sleep(10L);
+ return 1;
+ }
+ };
+ taskManager.setMainConsumer(consumer);
+
+ final StreamThread thread = new StreamThread(
+ mockTime,
+ config,
+ null,
+ consumer,
+ consumer,
+ changelogReader,
+ null,
+ taskManager,
+ streamsMetrics,
+ internalTopologyBuilder,
+ CLIENT_ID,
+ new LogContext(""),
+ new AtomicInteger(),
+ new AtomicLong(Long.MAX_VALUE)
+ );
+ thread.updateThreadMetadata("adminClientId");
+ thread.setState(StreamThread.State.STARTING);
+
+ final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
+ activeTasks.put(task1, Collections.singleton(t1p1));
+ thread.taskManager().handleAssignment(activeTasks, emptyMap());
+ thread.rebalanceListener().onPartitionsAssigned(Collections.singleton(t1p1));
+
+ assertTrue(
+ Double.isNaN(
+ (Double) streamsMetrics.metrics().get(new MetricName(
+ "commit-latency-max",
+ "stream-thread-metrics",
+ "",
+ Collections.singletonMap("thread-id", CLIENT_ID))
+ ).metricValue()
+ )
+ );
+ assertTrue(
+ Double.isNaN(
+ (Double) streamsMetrics.metrics().get(new MetricName(
+ "commit-latency-avg",
+ "stream-thread-metrics",
+ "",
+ Collections.singletonMap("thread-id", CLIENT_ID))
+ ).metricValue()
+ )
+ );
+
+ thread.runOnce();
+
+ assertThat(
+ streamsMetrics.metrics().get(
+ new MetricName(
+ "commit-latency-max",
+ "stream-thread-metrics",
+ "",
+ Collections.singletonMap("thread-id", CLIENT_ID)
+ )
+ ).metricValue(),
+ equalTo(10.0)
+ );
+ assertThat(
+ streamsMetrics.metrics().get(
+ new MetricName(
+ "commit-latency-avg",
+ "stream-thread-metrics",
+ "",
+ Collections.singletonMap("thread-id", CLIENT_ID)
+ )
+ ).metricValue(),
+ equalTo(10.0)
+ );
+ }
+
+ @Test
public void shouldInjectSharedProducerForAllTasksUsingClientSupplierOnCreateIfEosDisabled() {
internalTopologyBuilder.addSource(null, "source1", null, null, null, topic1);
internalStreamsBuilder.buildAndOptimizeTopology();
@@ -1234,14 +1338,11 @@ public class StreamThreadTest {
final AtomicBoolean processed = new AtomicBoolean(false);
internalTopologyBuilder.addProcessor(
"proc",
- () -> new Processor<Object, Object, Object, Object>() {
- @Override
- public void process(final Record<Object, Object> record) {
- if (shouldThrow.get()) {
- throw new TaskCorruptedException(singletonMap(task1, new HashSet<>(singleton(storeChangelogTopicPartition))));
- } else {
- processed.set(true);
- }
+ () -> record -> {
+ if (shouldThrow.get()) {
+ throw new TaskCorruptedException(singletonMap(task1, new HashSet<>(singleton(storeChangelogTopicPartition))));
+ } else {
+ processed.set(true);
}
},
"name");