jsancio commented on code in PR #18852:
URL: https://github.com/apache/kafka/pull/18852#discussion_r1968142451
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1802,10 +1804,31 @@ private boolean handleFetchResponse(
}
}
- private void appendAsFollower(
- Records records
- ) {
- LogAppendInfo info = log.appendAsFollower(records);
+ private static String convertToHexadecimal(Records records) {
+ ByteBuffer buffer = ((MemoryRecords) records).buffer();
+ byte[] bytes = new byte[Math.min(buffer.remaining(),
DefaultRecordBatch.RECORD_BATCH_OVERHEAD)];
+ buffer.get(bytes);
+
+ return HexFormat.of().formatHex(bytes);
+ }
+
+ private void appendAsFollower(Records records) {
+ if (records.sizeInBytes() == 0) {
+ // Nothing to do if there are no bytes in the response
+ return;
+ }
+
+ try {
+ var info = log.appendAsFollower(records, quorum.epoch());
+ kafkaRaftMetrics.updateFetchedRecords(info.lastOffset -
info.firstOffset + 1);
+ } catch (CorruptRecordException | InvalidRecordException e) {
+ logger.info(
Review Comment:
To me it is not technically an error. It is an expected state and handling
given the race between the network thread and the kraft driver thread.
There is nothing the user or operator can do to mitigate this condition and
state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
