Skip to content

MINOR: Log4j Improvements on Fetcher #8629

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

Merged
guozhangwang merged 3 commits into from
May 7, 2020
Merged

MINOR: Log4j Improvements on Fetcher #8629

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
859e03c
the fix
guozhangwang May 7, 2020
662364f
revert
guozhangwang May 7, 2020
eab85ef
minor fixes
guozhangwang May 7, 2020
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
8 changes: 5 additions & 3 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,7 +260,7 @@ public synchronized int sendFetches() {
}
RequestFuture<ClientResponse> future = client.send(fetchTarget, request);
// We add the node to the set of nodes with pending fetch requests before adding the
// listener because the future may have been fulfilled on another thread (e.g. during a
// listenerbecause the future may have been fulfilled on another thread (e.g. during a

@hachikuji hachikuji May 7, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I assume this was unintentional.

// disconnection being handled by the heartbeat thread) which will mean the listener
// will be invoked synchronously.
this.nodesWithPendingFetchRequests.add(entry.getKey().id());
Expand Down Expand Up @@ -676,13 +676,15 @@ private List<ConsumerRecord<K, V>> fetchRecords(CompletedFetch completedFetch, i
if (completedFetch.nextFetchOffset == position.offset) {
List<ConsumerRecord<K, V>> partRecords = completedFetch.fetchRecords(maxRecords);

log.trace("Returning fetched records {} at offset {} for assigned partition {}",
partRecords, position, completedFetch.partition);

@hachikuji hachikuji May 7, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I feel logging all of the records even at TRACE level will be too much. For example, our system tests often have TRACE enabled. Huge single-line log messages are difficult to consume both visually and in systems like elastic.

@guozhangwang guozhangwang May 7, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Makes sense, it is primarily for my local debugging of an integration test which only sends a couple records. I will only print the num.records instead.


if (completedFetch.nextFetchOffset > position.offset) {
SubscriptionState.FetchPosition nextPosition = new SubscriptionState.FetchPosition(
completedFetch.nextFetchOffset,
completedFetch.lastEpoch,
position.currentLeader);
log.trace("Returning fetched records at offset {} for assigned partition {} and update " +
"position to {}", position, completedFetch.partition, nextPosition);
log.trace("Update fetching position to {} for partition {}", nextPosition, completedFetch.partition);
subscriptions.position(completedFetch.partition, nextPosition);
}

Expand Down
2 changes: 2 additions & 0 deletions clients/src/main/java/org/apache/kafka/common/record/DefaultRecordBatch.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -484,6 +484,8 @@ static void writeHeader(ByteBuffer buffer,
@Override
public String toString() {
return "RecordBatch(magic=" + magic() + ", offsets=[" + baseOffset() + ", " + lastOffset() + "], " +
"sequence=[" + baseSequence() + ", " + lastSequence() + "], " +
"isTransactional=" + isTransactional() + ", isControlBatch=" + isControlBatch() + ", " +
"compression=" + compressionType() + ", timestampType=" + timestampType() + ", crc=" + checksum() + ")";
}

Expand Down