KAFKA-5490: Skip empty record batches in the consumer #3408
Conversation
ijuma
force-pushed
the
kafka-5490-consumer-should-skip-empty-batches
branch
from
b1121c4
to
991b51b
Compare
|
@dguy, can you please review this? |
|
Btw, this is @hachikuji's code, I just split the consumer change from the log cleaner changes. |
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
| @@ -1061,6 +1061,7 @@ private Record nextFetchedRecord() { | |||
| } | |||
|
|
|||
| records = currentBatch.streamingIterator(decompressionBufferSupplier); | |||
| continue; | |||
There was a problem hiding this comment.
could we use
} else {
Record record = records.next();
...
}
instead of continue here?
| @@ -393,6 +395,24 @@ private static byte computeAttributes(CompressionType type, TimestampType timest | |||
| return attributes; | |||
| } | |||
|
|
|||
| public static void writeEmptyHeader(ByteBuffer buffer, | |||
There was a problem hiding this comment.
If this is only for testing should we put it in a Test class?
There was a problem hiding this comment.
#3406 uses this in non-test code. We will merge that soon as well, so it seems like this way will cause less churn.
| @@ -1669,6 +1669,35 @@ protected boolean shouldRetain(RecordBatch recordBatch, Record record) { | |||
| } | |||
|
|
|||
| @Test | |||
| public void testUpdatePositionOnEmptyBatch() { | |||
There was a problem hiding this comment.
If writeEmptyHeader is only used by test code and is moved to a test utils class, then i guess we don't need this?
There was a problem hiding this comment.
As per my other comment, we hope to use it in non-test code soon.
There was a problem hiding this comment.
Hmm I think you meant to comment on testEmptyHeader. This test is used to validate that the fetcher change works correctly.
|
I updated the PR, @dguy. |
|
I will merge to trunk and 0.11.0 once the tests pass. |
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
The actual fix for KAFKA-5490 is in #3406. This is just the consumer change that will allow the cleaner to use empty record batches without breaking 0.11.0.0 consumers (assuming that KAFKA-5490 does not make the cut). This is a safe change even if we decide to go with a different option for KAFKA-5490 and I'd like to include it in RC2. Author: Jason Gustafson <jason@confluent.io> Author: Ismael Juma <ismael@juma.me.uk> Reviewers: Damian Guy <damian.guy@gmail.com>, Ismael Juma <ismael@juma.me.uk> Closes #3408 from ijuma/kafka-5490-consumer-should-skip-empty-batches (cherry picked from commit fc58ac5) Signed-off-by: Ismael Juma <ismael@juma.me.uk>
The actual fix for KAFKA-5490 is in
#3406.
This is just the consumer change that will allow the cleaner
to use empty record batches without breaking 0.11.0.0
consumers (assuming that KAFKA-5490 does not make the cut).
This is a safe change even if we decide to go with a different option
for KAFKA-5490 and I'd like to include it in RC2.