Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

KAFKA-13149: fix NPE for record==null when handling a produce request #11080

Merged
merged 3 commits into from
Sep 14, 2021
Merged

KAFKA-13149: fix NPE for record==null when handling a produce request #11080

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
14fa0cb
fix NPE when record==null in append
ccding Jul 15, 2021
b357b81
add test
ccding Jul 19, 2021
117c550
Merge branch 'trunk' into ak-record-null
ccding Sep 14, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion clients/src/main/java/org/apache/kafka/common/record/DefaultRecord.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -293,7 +293,9 @@ public static DefaultRecord readFrom(ByteBuffer buffer,
Long logAppendTime) {
int sizeOfBodyInBytes = ByteUtils.readVarint(buffer);
if (buffer.remaining() < sizeOfBodyInBytes)
return null;
throw new InvalidRecordException("Invalid record size: expected " + sizeOfBodyInBytes +
" bytes in record payload, but instead the buffer has only " + buffer.remaining() +
" remaining bytes.");
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this really an exceptional case? Don't we do reads where we don't know exactly where the read ends and hence will trigger this path?

Copy link
Contributor Author

@ccding ccding Jul 20, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are you saying the case that we are yet to complete reading the request? I didn't see a retry path, but it will cause a null point exception at

kafka/core/src/main/scala/kafka/log/LogValidator.scala

Line 191 in bfc57aa

if (!record.hasMagic(batch.magic)) {

What do you suggest I do here?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the intent here was to cover the case where an incomplete record is returned by the broker. However, we have broker logic to try and avoid this case since KIP-74:

} else if (!hardMaxBytesLimit && readInfo.fetchedData.firstEntryIncomplete) {
            // For FetchRequest version 3, we replace incomplete message sets with an empty one as consumers can make
            // progress in such cases and don't need to report a `RecordTooLargeException`
            FetchDataInfo(readInfo.fetchedData.fetchOffsetMetadata, MemoryRecords.EMPTY)

@hachikuji Do you remember if there is still a reason to return null here instead of the exception @ccding is proposing?

Copy link
Contributor Author

@ccding ccding Jul 20, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the case where an incomplete record is returned by the broker

I am referring to the produce API for the null pointer exception. The record is from a producer. The InvalidRecordException will trigger a response to the producer.

If the fetch path requires a different return value, I guess the problem becomes more complicated.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I understand you're talking about the producer case. I am talking about the fetch case. As I said, I think we may not need that special logic anymore, but @hachikuji would know for sure.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@hachikuji do you have time to have a look at this?

Copy link

@hachikuji hachikuji Sep 13, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Apologies for the delay here. I don't see a problem with the change. I believe that @ijuma is right that the fetch response may still return incomplete data, but I think this is handled in ByteBufferLogInputStream. We stop batch iteration early if there is incomplete data, so we would never reach the readFrom here which is called for each record in the batch. It's worth noting also that the only caller of this method (in DefaultRecordBatch.uncompressedIterator) has the following logic:

try {
  return DefaultRecord.readFrom(buffer, baseOffset, firstTimestamp, baseSequence, logAppendTime);
} catch (BufferUnderflowException e) {
  throw new InvalidRecordException("Incorrect declared batch size, premature EOF reached");
}

So it is already handle underflows in a similar way.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for checking @hachikuji.


int totalSizeInBytes = ByteUtils.sizeOfVarint(sizeOfBodyInBytes) + sizeOfBodyInBytes;
return readFrom(buffer, totalSizeInBytes, sizeOfBodyInBytes, baseOffset, baseTimestamp,
Expand Down
10 changes: 10 additions & 0 deletions clients/src/test/java/org/apache/kafka/common/record/DefaultRecordTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -481,4 +481,14 @@ public void testSerdeNoSequence() throws IOException {
assertEquals(RecordBatch.NO_SEQUENCE, record.sequence());
}

@Test
public void testInvalidSizeOfBodyInBytes() {
int sizeOfBodyInBytes = 10;
ByteBuffer buf = ByteBuffer.allocate(5);
ByteUtils.writeVarint(sizeOfBodyInBytes, buf);

buf.flip();
assertThrows(InvalidRecordException.class,
() -> DefaultRecord.readFrom(buf, 0L, 0L, RecordBatch.NO_SEQUENCE, null));
}
}