mimaison commented on code in PR #12545:
URL: https://github.com/apache/kafka/pull/12545#discussion_r1157554656
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java:
##########
@@ -1413,22 +1414,28 @@ private ConsumerRecord<K, V> parseRecord(TopicPartition
partition,
RecordBatch batch,
Record record) {
try {
- long offset = record.offset();
- long timestamp = record.timestamp();
- Optional<Integer> leaderEpoch =
maybeLeaderEpoch(batch.partitionLeaderEpoch());
- TimestampType timestampType = batch.timestampType();
- Headers headers = new RecordHeaders(record.headers());
- ByteBuffer keyBytes = record.key();
- byte[] keyByteArray = keyBytes == null ? null :
Utils.toArray(keyBytes);
- K key = keyBytes == null ? null :
this.keyDeserializer.deserialize(partition.topic(), headers, keyByteArray);
- ByteBuffer valueBytes = record.value();
- byte[] valueByteArray = valueBytes == null ? null :
Utils.toArray(valueBytes);
- V value = valueBytes == null ? null :
this.valueDeserializer.deserialize(partition.topic(), headers, valueByteArray);
- return new ConsumerRecord<>(partition.topic(),
partition.partition(), offset,
- timestamp, timestampType,
- keyByteArray == null ?
ConsumerRecord.NULL_SIZE : keyByteArray.length,
- valueByteArray == null ?
ConsumerRecord.NULL_SIZE : valueByteArray.length,
- key, value, headers, leaderEpoch);
+ final long offset = record.offset();
Review Comment:
There are similar changes in some of the other files too.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java:
##########
@@ -1413,22 +1414,28 @@ private ConsumerRecord<K, V> parseRecord(TopicPartition
partition,
RecordBatch batch,
Record record) {
try {
- long offset = record.offset();
- long timestamp = record.timestamp();
- Optional<Integer> leaderEpoch =
maybeLeaderEpoch(batch.partitionLeaderEpoch());
- TimestampType timestampType = batch.timestampType();
- Headers headers = new RecordHeaders(record.headers());
- ByteBuffer keyBytes = record.key();
- byte[] keyByteArray = keyBytes == null ? null :
Utils.toArray(keyBytes);
- K key = keyBytes == null ? null :
this.keyDeserializer.deserialize(partition.topic(), headers, keyByteArray);
- ByteBuffer valueBytes = record.value();
- byte[] valueByteArray = valueBytes == null ? null :
Utils.toArray(valueBytes);
- V value = valueBytes == null ? null :
this.valueDeserializer.deserialize(partition.topic(), headers, valueByteArray);
- return new ConsumerRecord<>(partition.topic(),
partition.partition(), offset,
- timestamp, timestampType,
- keyByteArray == null ?
ConsumerRecord.NULL_SIZE : keyByteArray.length,
- valueByteArray == null ?
ConsumerRecord.NULL_SIZE : valueByteArray.length,
- key, value, headers, leaderEpoch);
+ final long offset = record.offset();
Review Comment:
We typically don't make local variables `final` in Kafka (apart from
Streams), so could we undo these changes?
--
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
