KAFKA-4429; records-lag should be zero if FetchResponse is empty #2155

lindong28 · 2016-11-22T00:39:35Z

No description provided.

hachikuji · 2016-11-28T18:18:06Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java

@@ -689,6 +689,9 @@ private void handleListOffsetResponse(Map<TopicPartition, Long> timestampsToSear
                    parsedRecords = new PartitionRecords<>(fetchOffset, tp, parsed);
                    ConsumerRecord<K, V> record = parsed.get(parsed.size() - 1);
                    this.sensors.recordsFetchLag.record(partition.highWatermark - record.offset());
+                } else if (partition.records.iterator().hasNext()) {


This doesn't seem right. At a minimum, we're missing the '!' to invert the check. Additionally, iteration through a record set is not cheap (it may involve decompression), so we should try to use the work we've already done. In fact, maybe we could remove the check and compute the lag based on the high watermark and the current position of the consumer instead?

@hachikuji Yes, good point. I think we can compute lag as partition.highWatermark - position if parsed result is empty. But I am not sure if we should use this as lag if parsed result is not empty. It may be a good solution. But doing so changes the semantic of the fetch lag metric since the metric would only be updated when position is updated in the next poll() call.

Good point. This is a little subtle, but it doesn't seem incorrect. Since we know that the fetched offset matches our current position, the fetch lag should be in sync with the position. In fact, it may be more intuitive if we use fetchOffset instead of position.

@hachikuji Sorry for late reply. Sure, I will replace position with fetchOffset in the patch. I still keep the existing measurement of fetchLag in the case where the response is not empty because I don't want to cause any unexpected problem by making this subtle but backward incompatible change. On the other hand I don't mind if you or someone else want to change it.

ewencp · 2017-01-02T19:41:06Z

@lindong28 FetcherTest has a test or two that do validation of metrics, might be worth adding a few assertions to validate this change (and maybe a test if both lag and non-lag conditions aren't available in existing tests).

asfbot · 2017-01-02T20:27:20Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/421/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-01-02T21:19:19Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/423/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-02T21:35:40Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/422/
Test FAILed (JDK 8 and Scala 2.12).

lindong28 · 2017-01-03T00:41:49Z

@ewencp Sure, I just added a unit test for this patch. Can you take another look? Thank you.

asfbot · 2017-01-03T01:27:47Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/428/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-01-03T02:16:46Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/429/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-03T02:19:41Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/430/
Test FAILed (JDK 8 and Scala 2.11).

hachikuji · 2017-01-03T17:47:47Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java

+    }
+
+    private Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> fetchRecords(MemoryRecords records, short error, long hw, int throttleTime) {
+        this.records = records;


Is this needed?

Thanks. I copied & pasted this logic without checking if it is needed. Fixed now.

hachikuji · 2017-01-03T17:47:59Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java

+            MemoryRecordsBuilder builder = MemoryRecords.builder(ByteBuffer.allocate(1024), CompressionType.NONE, TimestampType.CREATE_TIME);
+            for (int v = 0; v < 3; v++)
+                builder.appendWithOffset((long) i * 3 + v, Record.NO_TIMESTAMP, "key".getBytes(), String.format("value-%d", v).getBytes());
+            records = fetchRecords(builder.build(), Errors.NONE.code(), 100L, 100 * i).get(tp);


nit: could we use a local variable instead of overriding the field?

Sorry, records is already a variable, but it seems we could move its initialization to here?

Yes we can. Personally I think the logic is clearer by passing it as parameter of fetchRecords() instead of via the class variable. Do you think the updated patch looks better?

To clarify, what I meant is just to move the initialization of the records variable from above to here (since we don't actually need it outside of the loop):

List<ConsumerRecord<byte[], byte[]>> records = fetchRecords(builder.build(), Errors.NONE.code(), 100L, 100 * i).get(tp);

Ah I see. Sure, it is fixed now.

hachikuji · 2017-01-03T18:39:56Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java

+
+        // recordsFetchLagMax should be hw - fetchOffset after receiving an empty FetchResponse
+        MemoryRecordsBuilder builder = MemoryRecords.builder(ByteBuffer.allocate(1024), CompressionType.NONE, TimestampType.CREATE_TIME);
+        fetchRecords(builder.build(), Errors.NONE.code(), 100L, 0);


You could also use MemoryRecords.EMPTY here (I think).

Nice point. It works:)

asfbot · 2017-01-03T20:29:48Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/443/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-01-03T20:31:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/444/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-03T20:35:39Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/442/
Test FAILed (JDK 7 and Scala 2.10).

hachikuji · 2017-01-03T21:50:58Z

LGTM. Thanks for the patch!

asfbot · 2017-01-03T22:04:00Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/448/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-01-03T22:57:10Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/450/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-03T23:56:27Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/449/
Test FAILed (JDK 8 and Scala 2.12).

Author: Dong Lin <lindong28@gmail.com> Reviewers: Ewen Cheslack-Postava <me@ewencp.org>, Jason Gustafson <jason@confluent.io> Closes apache#2155 from lindong28/KAFKA-4429

hachikuji reviewed Nov 28, 2016

View reviewed changes

lindong28 added 3 commits January 2, 2017 14:46

KAFKA-4429; records-lag should be zero if FetchResponse is empty

f6a88cb

Address comment

4bc5475

Add unit test

4d6de50

lindong28 force-pushed the KAFKA-4429 branch from d731473 to 4d6de50 Compare January 3, 2017 00:41

hachikuji reviewed Jan 3, 2017

View reviewed changes

lindong28 added 2 commits January 3, 2017 10:07

Address review

f4f81b8

Address review

3d7a86b

hachikuji reviewed Jan 3, 2017

View reviewed changes

Address review

2420ff0

asfgit closed this in b565dd7 Jan 3, 2017

lindong28 deleted the KAFKA-4429 branch January 5, 2017 21:41

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-4429; records-lag should be zero if FetchResponse is empty #2155

KAFKA-4429; records-lag should be zero if FetchResponse is empty #2155

lindong28 commented Nov 22, 2016

hachikuji Nov 28, 2016

lindong28 Nov 30, 2016

hachikuji Nov 30, 2016 •

edited

Loading

lindong28 Jan 2, 2017

ewencp commented Jan 2, 2017

asfbot commented Jan 2, 2017

asfbot commented Jan 2, 2017

asfbot commented Jan 2, 2017

lindong28 commented Jan 3, 2017

asfbot commented Jan 3, 2017

asfbot commented Jan 3, 2017

asfbot commented Jan 3, 2017

hachikuji Jan 3, 2017

lindong28 Jan 3, 2017

hachikuji Jan 3, 2017

hachikuji Jan 3, 2017

lindong28 Jan 3, 2017

hachikuji Jan 3, 2017

lindong28 Jan 3, 2017

hachikuji Jan 3, 2017

lindong28 Jan 3, 2017

asfbot commented Jan 3, 2017

asfbot commented Jan 3, 2017

asfbot commented Jan 3, 2017

hachikuji commented Jan 3, 2017

asfbot commented Jan 3, 2017

asfbot commented Jan 3, 2017

asfbot commented Jan 3, 2017

KAFKA-4429; records-lag should be zero if FetchResponse is empty #2155

KAFKA-4429; records-lag should be zero if FetchResponse is empty #2155

Conversation

lindong28 commented Nov 22, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji Nov 30, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ewencp commented Jan 2, 2017

asfbot commented Jan 2, 2017

asfbot commented Jan 2, 2017

asfbot commented Jan 2, 2017

lindong28 commented Jan 3, 2017

asfbot commented Jan 3, 2017

asfbot commented Jan 3, 2017

asfbot commented Jan 3, 2017

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

asfbot commented Jan 3, 2017

asfbot commented Jan 3, 2017

asfbot commented Jan 3, 2017

hachikuji commented Jan 3, 2017

asfbot commented Jan 3, 2017

asfbot commented Jan 3, 2017

asfbot commented Jan 3, 2017

hachikuji Nov 30, 2016 •

edited

Loading