[ 
https://issues.apache.org/jira/browse/KAFKA-21152?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Eric Chang updated KAFKA-21152:
-------------------------------
    Description: 
h3. What happened

{{LogCompactionTest.test_log_compaction}} is flaky. In the 4.4.0-rc1 system 
test run (zstd level 10, ISOLATED_KRAFT), the tester read only 7498 of about 
191k rows and the validation failed. The same case passed 5 out of 5 times on 
rerun. The trimmed logs are attached.

The cause is in the tester, not in the broker or the consumer. 
{{LogCompactionTester.consumeMessages()}} 
({{tools/src/main/java/org/apache/kafka/tools/LogCompactionTester.java}}) stops 
at the first empty {{poll()}}:

{code:java}
while (true) {
    ConsumerRecords<String, String> consumerRecords = 
consumer.poll(Duration.ofSeconds(20));
    if (consumerRecords.isEmpty()) return consumedFilePath;
    ...
}
{code}

Since KAFKA-12980 (3.2.0), {{poll()}} can return no records even when more data 
remains. So {{consumeMessages()}} sometimes stops early, the consumed file is 
incomplete, and {{validateOutput()}} fails. This is not a 4.4 regression: the 
consumer behavior exists since 3.2.0, and the tester code is even older.

h3. Details

Why {{poll()}} returns no records: after compaction, a batch keeps its 
{{lastOffset}} even when its last records are removed. If one {{poll()}} 
returns exactly {{max.poll.records}} records and the last one is the last 
retained record of a fetch, the next {{poll()}} only moves the position to 
{{lastOffset + 1}} and returns right away with no records. In the failed run, 
the tester stopped at a batch with {{lastOffset}} 649433 whose last retained 
record is 649426. The consumer behavior is expected (KAFKA-14865 tracks it). 
This ticket only fixes the tester.

Simple repro (offsets 5..9 compacted away; {{fetch.max.bytes=1}} for one batch 
per fetch, {{max.poll.records=5}}):

{noformat}
poll#1 count=5 position=5  endOffset=16   (offsets 0..4)
poll#2 count=0 position=10 endOffset=16   (skips removed offsets 5..9: empty, 
but not at the end)
poll#3 count=5 position=15 endOffset=16   (offsets 10..14)
{noformat}

Fix: in {{consumeMessages()}}, get the end offsets before consuming, and poll 
until the position of every partition reaches its end offset. Give up only when 
a poll returns no records and moves no position.

  was:
{{LogCompactionTest.test_log_compaction}} is flaky because 
{{LogCompactionTester.consumeMessages()}} 
({{tools/src/main/java/org/apache/kafka/tools/LogCompactionTester.java}}) stops 
at the first empty {{poll()}}:

{code:java}
while (true) {
    ConsumerRecords<String, String> consumerRecords = 
consumer.poll(Duration.ofSeconds(20));
    if (consumerRecords.isEmpty()) return consumedFilePath;
    ...
}
{code}

Since KAFKA-12980 (3.2.0), {{poll()}} can return no records even when more data 
remains. So {{consumeMessages()}} sometimes stops early, the consumed file is 
incomplete, and {{validateOutput()}} fails. This is a tester bug, not a 4.4 
regression: the consumer behavior exists since 3.2.0, and the tester code is 
even older.

Why {{poll()}} returns no records: after compaction, a batch keeps its 
{{lastOffset}} even when its last records are removed. If one {{poll()}} 
returns exactly {{max.poll.records}} records and the last one is the last live 
record of a fetch, the next {{poll()}} only moves the position to {{lastOffset 
+ 1}} and returns right away with no records. The consumer behavior is expected 
(KAFKA-14865 tracks it). This ticket only fixes the tester.

Simple repro (offsets 5..9 compacted away; {{fetch.max.bytes=1}} for one batch 
per fetch, {{max.poll.records=5}}):

{noformat}
poll#1 count=5 position=5  endOffset=16   (offsets 0..4)
poll#2 count=0 position=10 endOffset=16   (skips removed offsets 5..9: empty, 
but not at the end)
poll#3 count=5 position=15 endOffset=16   (offsets 10..14)
{noformat}

Evidence: in the 4.4.0-rc1 system test run (zstd level 10, ISOLATED_KRAFT), the 
tester read only 7498 of about 191k rows. It stopped at a batch with 
{{lastOffset}} 649433 whose last live record is 649426. The same case passed 5 
out of 5 times on rerun. The trimmed logs are attached.

Fix: in {{consumeMessages()}}, get the end offsets before consuming, and poll 
until the position of every partition reaches its end offset, with an overall 
timeout.


> LogCompactionTester should read until the end offsets instead of stopping at 
> an empty poll
> ------------------------------------------------------------------------------------------
>
>                 Key: KAFKA-21152
>                 URL: https://issues.apache.org/jira/browse/KAFKA-21152
>             Project: Kafka
>          Issue Type: Bug
>          Components: system tests, tools
>            Reporter: Eric Chang
>            Assignee: Eric Chang
>            Priority: Minor
>              Labels: flaky-test
>         Attachments: build948-trimmed.log
>
>
> h3. What happened
> {{LogCompactionTest.test_log_compaction}} is flaky. In the 4.4.0-rc1 system 
> test run (zstd level 10, ISOLATED_KRAFT), the tester read only 7498 of about 
> 191k rows and the validation failed. The same case passed 5 out of 5 times on 
> rerun. The trimmed logs are attached.
> The cause is in the tester, not in the broker or the consumer. 
> {{LogCompactionTester.consumeMessages()}} 
> ({{tools/src/main/java/org/apache/kafka/tools/LogCompactionTester.java}}) 
> stops at the first empty {{poll()}}:
> {code:java}
> while (true) {
>     ConsumerRecords<String, String> consumerRecords = 
> consumer.poll(Duration.ofSeconds(20));
>     if (consumerRecords.isEmpty()) return consumedFilePath;
>     ...
> }
> {code}
> Since KAFKA-12980 (3.2.0), {{poll()}} can return no records even when more 
> data remains. So {{consumeMessages()}} sometimes stops early, the consumed 
> file is incomplete, and {{validateOutput()}} fails. This is not a 4.4 
> regression: the consumer behavior exists since 3.2.0, and the tester code is 
> even older.
> h3. Details
> Why {{poll()}} returns no records: after compaction, a batch keeps its 
> {{lastOffset}} even when its last records are removed. If one {{poll()}} 
> returns exactly {{max.poll.records}} records and the last one is the last 
> retained record of a fetch, the next {{poll()}} only moves the position to 
> {{lastOffset + 1}} and returns right away with no records. In the failed run, 
> the tester stopped at a batch with {{lastOffset}} 649433 whose last retained 
> record is 649426. The consumer behavior is expected (KAFKA-14865 tracks it). 
> This ticket only fixes the tester.
> Simple repro (offsets 5..9 compacted away; {{fetch.max.bytes=1}} for one 
> batch per fetch, {{max.poll.records=5}}):
> {noformat}
> poll#1 count=5 position=5  endOffset=16   (offsets 0..4)
> poll#2 count=0 position=10 endOffset=16   (skips removed offsets 5..9: empty, 
> but not at the end)
> poll#3 count=5 position=15 endOffset=16   (offsets 10..14)
> {noformat}
> Fix: in {{consumeMessages()}}, get the end offsets before consuming, and poll 
> until the position of every partition reaches its end offset. Give up only 
> when a poll returns no records and moves no position.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

Reply via email to