KAFKA-9753: Add active tasks process ratio #8370

guozhangwang · 2020-03-27T00:30:02Z

Measure the percentage ratio the stream thread spent on processing each task among all assigned active tasks (KIP-444).

Also add unit tests to cover the added metrics in this PR and the previous #8358

Also trying to fix the flaky test reported in KAFKA-5842

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

…op-tasks-order

…tive-process-ratio

…op-tasks-order

guozhangwang

Only be reviewed after #8358

guozhangwang · 2020-03-27T00:33:17Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java

@@ -192,6 +192,10 @@ public boolean isValidTransition(final State newState) {

    void markChangelogAsCorrupted(final Collection<TopicPartition> partitions);

+    void recordProcessLatency(final long elapsedLatencyMs);


This is a bit awkward.. if anyone have a more elegant solution I'm all ears.

prop: You could record the time spent in processing in Task#process() and just have a method Task#recordProcessRatio(final long totalProcessTime) in the interface.

Hey @guozhangwang , I did have some ideas (similar to @cadonna 's), but it might be easier to just present it as a diff: guozhangwang#12

I originally did this, but I'm concerned that the latency for processing a single record is sub-milliseconds and hence the cumulated latency would be zero.

guozhangwang · 2020-03-27T06:02:33Z

call @vvcephei @cadonna for reviews

…tive-process-ratio

guozhangwang · 2020-03-27T21:44:30Z

test this please

guozhangwang · 2020-03-27T21:44:41Z

test this please

guozhangwang

Fixed unit tests, cc @vvcephei

guozhangwang · 2020-03-27T21:47:56Z

streams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java

@@ -646,8 +645,8 @@ public void concurrentAccesses() throws Exception {
        producerThread.start();

        try {
-            waitUntilAtLeastNumRecordProcessed(outputTopicConcurrent, numberOfWordsPerIteration);
-            waitUntilAtLeastNumRecordProcessed(outputTopicConcurrentWindowed, numberOfWordsPerIteration);
+            waitUntilAtLeastNumRecordProcessed(outputTopicConcurrent, 1);


This test becomes more flaky than before after the main loop refactoring, shown as the following:

java.lang.AssertionError: Did not receive all 48 records from topic output-concurrent-2 within 120000 ms Expected: is a value equal to or greater than <48> but: <38> was less than <48>

Each iteration has 48 words, and 38 unique words, and when it fails it reported only 38 records ever consumed from the output topic. I suspect this is because of the caching effects, we processed all 500000 iterations without sending out any updates to the resulted count table output topic (seems we process too fast?).

I piggy-backed a fix for this flakiness by reducing the requirement that we start the verification as long as we've seen one record in the output, similar as we do in other tests, and then in verifyGreaterOrEqual we do not require all keys been populated already.

prop: I would prefer to have this fix in its own PR, because the commit would be easier to retrieve and to reference on failure tickets.

guozhangwang · 2020-03-27T21:48:48Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

+
+            // we record the ratio out of the while loop so that the accumulated latency spans over
+            // multiple iterations with reasonably large max.num.records and hence is less vulnerable to outliers
+            taskManager.recordTaskProcessRatio(totalProcessLatency);


Note we should only call this inside the if (state == State.RUNNING) condition otherwise we may get ConcurrentModification exception.

Q: Could you elaborate on this?

I do not know exactly what's happening here, my guess is that in PARTITION_REVOKED state the tasks might be modified.

guozhangwang · 2020-03-30T22:28:29Z

test this please

cadonna

@guozhangwang Thank you for the PR!

Here my feedback

cadonna · 2020-03-31T16:16:38Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/TaskMetrics.java

@@ -96,6 +103,22 @@ public static Sensor processLatencySensor(final String threadId,
        return emptySensor(threadId, taskId, PROCESS_LATENCY, RecordingLevel.DEBUG, streamsMetrics);
    }

+    public static Sensor activeProcessRatioSensor(final String threadId,


req: Could you please add tests for this new sensor to TaskMetricsTest.

cadonna · 2020-03-31T16:30:36Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/TaskMetrics.java

+        sensor.add(
+            new MetricName(
+                name,
+                TASK_LEVEL_GROUP,
+                PROCESS_RATIO_DESCRIPTION,
+                streamsMetrics.taskLevelTagMap(threadId, taskId)),
+            new Value()
+        );


Suggested change

sensor.add(

new MetricName(

name,

TASK_LEVEL_GROUP,

PROCESS_RATIO_DESCRIPTION,

streamsMetrics.taskLevelTagMap(threadId, taskId)),

new Value()

);

addValueMetricToSensor(

sensor,

TASK_LEVEL_GROUP,

streamsMetrics.taskLevelTagMap(threadId, taskId),

name,

PROCESS_RATIO_DESCRIPTION

);

req: Please use StreamsMetricsImpl#addValueMetricToSensor()

cadonna · 2020-03-31T18:15:51Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java

@@ -68,6 +70,17 @@ public void markChangelogAsCorrupted(final Collection<TopicPartition> partitions
        stateMgr.markChangelogAsCorrupted(partitions);
    }

+    @Override
+    public void recordProcessLatency(final long elapsedLatencyMs) {


prop: I find processLatency and totalLatency a bit misleading. What about processTime and totalProcessTime.

cadonna · 2020-03-31T18:21:20Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

@@ -616,6 +619,12 @@ public boolean process(final long wallClockTime) {
        return true;
    }

+    @Override
+    public void recordTotalLatency(final long elapsedLatencyMs) {


prop: Could you rename recordTotalLatency() to recordProcessRatioAndReset().

cadonna · 2020-03-31T18:24:13Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

+
+            // we record the ratio out of the while loop so that the accumulated latency spans over
+            // multiple iterations with reasonably large max.num.records and hence is less vulnerable to outliers
+            taskManager.recordTaskProcessRatio(totalProcessLatency);


Q: Could you elaborate on this?

cadonna · 2020-03-31T18:29:58Z

streams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java

@@ -646,8 +645,8 @@ public void concurrentAccesses() throws Exception {
        producerThread.start();

        try {
-            waitUntilAtLeastNumRecordProcessed(outputTopicConcurrent, numberOfWordsPerIteration);
-            waitUntilAtLeastNumRecordProcessed(outputTopicConcurrentWindowed, numberOfWordsPerIteration);
+            waitUntilAtLeastNumRecordProcessed(outputTopicConcurrent, 1);


prop: I would prefer to have this fix in its own PR, because the commit would be easier to retrieve and to reference on failure tickets.

cadonna · 2020-03-31T18:45:47Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java

@@ -192,6 +192,10 @@ public boolean isValidTransition(final State newState) {

    void markChangelogAsCorrupted(final Collection<TopicPartition> partitions);

+    void recordProcessLatency(final long elapsedLatencyMs);


prop: You could record the time spent in processing in Task#process() and just have a method Task#recordProcessRatio(final long totalProcessTime) in the interface.

cadonna · 2020-03-31T18:50:51Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

@@ -616,6 +619,12 @@ public boolean process(final long wallClockTime) {
        return true;
    }

+    @Override
+    public void recordTotalLatency(final long elapsedLatencyMs) {


req: Could you please add a unit test for this method?

cadonna · 2020-03-31T18:51:07Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java

@@ -68,6 +70,17 @@ public void markChangelogAsCorrupted(final Collection<TopicPartition> partitions
        stateMgr.markChangelogAsCorrupted(partitions);
    }

+    @Override
+    public void recordProcessLatency(final long elapsedLatencyMs) {


req: Could you please add a unit test for this method?

…nals/StreamTask.java

Some suggestions

vvcephei

LGTM, modulo the remainder of @cadonna 's feedback.

…tive-process-ratio

…angwang/kafka into K9753-active-process-ratio

guozhangwang added 13 commits March 24, 2020 17:43

first pass

70ca574

Merge branch 'trunk' of https://github.com/apache/kafka into K9756-lo…

8b33c3e

…op-tasks-order

fix unit tests

77e475e

add unit tests

d5c743b

Merge branch 'trunk' of https://github.com/apache/kafka into K9753-ac…

e81f042

…tive-process-ratio

first pass

8eb82b2

Merge branch 'trunk' of https://github.com/apache/kafka into K9756-lo…

b298236

…op-tasks-order

should be INFO

153b741

fix unit tests

a7d945c

Merge branch 'trunk' of https://github.com/apache/kafka into K9756-lo…

7431c1a

…op-tasks-order

github comments

0286037

rebase

bc5010e

unit tests

decaa94

guozhangwang commented Mar 27, 2020

View reviewed changes

guozhangwang mentioned this pull request Mar 27, 2020

KAFKA-9753: A few more metrics to add #8371

Merged

3 tasks

rebase from trunk

5bbf703

vvcephei added the streams label Mar 27, 2020

vvcephei self-requested a review March 27, 2020 18:27

guozhangwang added 2 commits March 27, 2020 12:03

Merge branch 'trunk' of https://github.com/apache/kafka into K9753-ac…

8023e10

…tive-process-ratio

fix unit tests

f2768e9

guozhangwang commented Mar 27, 2020

View reviewed changes

cadonna reviewed Mar 31, 2020

View reviewed changes

Some suggestions

7fbe8e0

vvcephei mentioned this pull request Mar 31, 2020

Some suggestions guozhangwang/kafka#12

Merged

3 tasks

John Roesler and others added 2 commits March 31, 2020 16:14

remove part of the suggestion

72c8bc2

Update streams/src/main/java/org/apache/kafka/streams/processor/inter…

3fc5d22

…nals/StreamTask.java

vvcephei and others added 2 commits March 31, 2020 16:20

Update streams/src/main/java/org/apache/kafka/streams/processor/inter…

5e2557c

…nals/StreamTask.java

Merge pull request #12 from vvcephei/john-8370-suggestion

cc224e9

Some suggestions

vvcephei approved these changes Mar 31, 2020

View reviewed changes

guozhangwang added 3 commits March 31, 2020 15:38

Merge branch 'trunk' of https://github.com/apache/kafka into K9753-ac…

15a15cc

…tive-process-ratio

Merge branch 'K9753-active-process-ratio' of https://github.com/guozh…

707ec0d

…angwang/kafka into K9753-active-process-ratio

github comments

32ef38c

guozhangwang merged commit 353aa62 into apache:trunk Mar 31, 2020

guozhangwang deleted the K9753-active-process-ratio branch March 31, 2020 23:39

mjsax added the kip Requires or implements a KIP label Jun 12, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-9753: Add active tasks process ratio #8370

KAFKA-9753: Add active tasks process ratio #8370

guozhangwang commented Mar 27, 2020 •

edited

guozhangwang left a comment

guozhangwang Mar 27, 2020

cadonna Mar 31, 2020

vvcephei Mar 31, 2020

guozhangwang Mar 31, 2020

guozhangwang commented Mar 27, 2020

guozhangwang commented Mar 27, 2020

guozhangwang commented Mar 27, 2020

guozhangwang left a comment

guozhangwang Mar 27, 2020

cadonna Mar 31, 2020

guozhangwang Mar 27, 2020

cadonna Mar 31, 2020

guozhangwang Mar 31, 2020

guozhangwang commented Mar 30, 2020

cadonna left a comment

cadonna Mar 31, 2020

guozhangwang Mar 31, 2020

cadonna Mar 31, 2020

guozhangwang Mar 31, 2020

cadonna Mar 31, 2020

guozhangwang Mar 31, 2020

cadonna Mar 31, 2020

cadonna Mar 31, 2020

cadonna Mar 31, 2020

cadonna Mar 31, 2020

cadonna Mar 31, 2020

cadonna Mar 31, 2020

vvcephei left a comment

		@@ -192,6 +192,10 @@ public boolean isValidTransition(final State newState) {

		void markChangelogAsCorrupted(final Collection<TopicPartition> partitions);

		void recordProcessLatency(final long elapsedLatencyMs);

KAFKA-9753: Add active tasks process ratio #8370

KAFKA-9753: Add active tasks process ratio #8370

Conversation

guozhangwang commented Mar 27, 2020 • edited

Committer Checklist (excluded from commit message)

guozhangwang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang commented Mar 27, 2020

guozhangwang commented Mar 27, 2020

guozhangwang commented Mar 27, 2020

guozhangwang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang commented Mar 30, 2020

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei left a comment

Choose a reason for hiding this comment

guozhangwang commented Mar 27, 2020 •

edited