diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml index f45e362181aa0..d3fc70af9eb12 100644 --- a/checkstyle/import-control.xml +++ b/checkstyle/import-control.xml @@ -119,6 +119,7 @@ + diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml index 7b5ae4009e581..f6bafbfcfe56f 100644 --- a/checkstyle/suppressions.xml +++ b/checkstyle/suppressions.xml @@ -171,6 +171,8 @@ + + diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 3251f4fa804e8..ed14c97712423 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -310,7 +310,7 @@ public void onSuccess(ClientResponse resp) { log.debug("Fetch {} at offset {} for partition {} returned fetch data {}", isolationLevel, fetchOffset, partition, partitionData); - Iterator batches = partitionData.records.batches().iterator(); + Iterator batches = partitionData.records().batches().iterator(); short responseVersion = resp.requestHeader().apiVersion(); completedFetches.add(new CompletedFetch(partition, partitionData, @@ -616,7 +616,7 @@ public Map>> fetchedRecords() { // in cases such as the TopicAuthorizationException, and the second condition ensures that no // potential data loss due to an exception in a following record. FetchResponse.PartitionData partition = records.partitionData; - if (fetched.isEmpty() && (partition.records == null || partition.records.sizeInBytes() == 0)) { + if (fetched.isEmpty() && (partition.records() == null || partition.records().sizeInBytes() == 0)) { completedFetches.poll(); } throw e; @@ -1210,7 +1210,7 @@ private CompletedFetch initializeCompletedFetch(CompletedFetch nextCompletedFetc FetchResponse.PartitionData partition = nextCompletedFetch.partitionData; long fetchOffset = nextCompletedFetch.nextFetchOffset; CompletedFetch completedFetch = null; - Errors error = partition.error; + Errors error = partition.error(); try { if (!subscriptions.hasValidPosition(tp)) { @@ -1227,11 +1227,11 @@ private CompletedFetch initializeCompletedFetch(CompletedFetch nextCompletedFetc } log.trace("Preparing to read {} bytes of data for partition {} with offset {}", - partition.records.sizeInBytes(), tp, position); - Iterator batches = partition.records.batches().iterator(); + partition.records().sizeInBytes(), tp, position); + Iterator batches = partition.records().batches().iterator(); completedFetch = nextCompletedFetch; - if (!batches.hasNext() && partition.records.sizeInBytes() > 0) { + if (!batches.hasNext() && partition.records().sizeInBytes() > 0) { if (completedFetch.responseVersion < 3) { // Implement the pre KIP-74 behavior of throwing a RecordTooLargeException. Map recordTooLargePartitions = Collections.singletonMap(tp, fetchOffset); @@ -1249,26 +1249,26 @@ private CompletedFetch initializeCompletedFetch(CompletedFetch nextCompletedFetc } } - if (partition.highWatermark >= 0) { - log.trace("Updating high watermark for partition {} to {}", tp, partition.highWatermark); - subscriptions.updateHighWatermark(tp, partition.highWatermark); + if (partition.highWatermark() >= 0) { + log.trace("Updating high watermark for partition {} to {}", tp, partition.highWatermark()); + subscriptions.updateHighWatermark(tp, partition.highWatermark()); } - if (partition.logStartOffset >= 0) { - log.trace("Updating log start offset for partition {} to {}", tp, partition.logStartOffset); - subscriptions.updateLogStartOffset(tp, partition.logStartOffset); + if (partition.logStartOffset() >= 0) { + log.trace("Updating log start offset for partition {} to {}", tp, partition.logStartOffset()); + subscriptions.updateLogStartOffset(tp, partition.logStartOffset()); } - if (partition.lastStableOffset >= 0) { - log.trace("Updating last stable offset for partition {} to {}", tp, partition.lastStableOffset); - subscriptions.updateLastStableOffset(tp, partition.lastStableOffset); + if (partition.lastStableOffset() >= 0) { + log.trace("Updating last stable offset for partition {} to {}", tp, partition.lastStableOffset()); + subscriptions.updateLastStableOffset(tp, partition.lastStableOffset()); } - if (partition.preferredReadReplica.isPresent()) { - subscriptions.updatePreferredReadReplica(completedFetch.partition, partition.preferredReadReplica.get(), () -> { + if (partition.preferredReadReplica().isPresent()) { + subscriptions.updatePreferredReadReplica(completedFetch.partition, partition.preferredReadReplica().get(), () -> { long expireTimeMs = time.milliseconds() + metadata.metadataExpireMs(); log.debug("Updating preferred read replica for partition {} to {}, set to expire at {}", - tp, partition.preferredReadReplica.get(), expireTimeMs); + tp, partition.preferredReadReplica().get(), expireTimeMs); return expireTimeMs; }); } @@ -1630,13 +1630,13 @@ private boolean isBatchAborted(RecordBatch batch) { } private PriorityQueue abortedTransactions(FetchResponse.PartitionData partition) { - if (partition.abortedTransactions == null || partition.abortedTransactions.isEmpty()) + if (partition.abortedTransactions() == null || partition.abortedTransactions().isEmpty()) return null; PriorityQueue abortedTransactions = new PriorityQueue<>( - partition.abortedTransactions.size(), Comparator.comparingLong(o -> o.firstOffset) + partition.abortedTransactions().size(), Comparator.comparingLong(o -> o.firstOffset) ); - abortedTransactions.addAll(partition.abortedTransactions); + abortedTransactions.addAll(partition.abortedTransactions()); return abortedTransactions; } diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java index 670de3755764a..2a4ffa8a70d40 100644 --- a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java +++ b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java @@ -67,6 +67,8 @@ import org.apache.kafka.common.message.EndTxnResponseData; import org.apache.kafka.common.message.ExpireDelegationTokenRequestData; import org.apache.kafka.common.message.ExpireDelegationTokenResponseData; +import org.apache.kafka.common.message.FetchRequestData; +import org.apache.kafka.common.message.FetchResponseData; import org.apache.kafka.common.message.FindCoordinatorRequestData; import org.apache.kafka.common.message.FindCoordinatorResponseData; import org.apache.kafka.common.message.HeartbeatRequestData; @@ -114,8 +116,6 @@ import org.apache.kafka.common.protocol.types.Struct; import org.apache.kafka.common.protocol.types.Type; import org.apache.kafka.common.record.RecordBatch; -import org.apache.kafka.common.requests.FetchRequest; -import org.apache.kafka.common.requests.FetchResponse; import org.apache.kafka.common.requests.ListOffsetRequest; import org.apache.kafka.common.requests.ListOffsetResponse; import org.apache.kafka.common.requests.OffsetsForLeaderEpochRequest; @@ -137,7 +137,7 @@ */ public enum ApiKeys { PRODUCE(0, "Produce", ProduceRequest.schemaVersions(), ProduceResponse.schemaVersions()), - FETCH(1, "Fetch", FetchRequest.schemaVersions(), FetchResponse.schemaVersions()), + FETCH(1, "Fetch", FetchRequestData.SCHEMAS, FetchResponseData.SCHEMAS), LIST_OFFSETS(2, "ListOffsets", ListOffsetRequest.schemaVersions(), ListOffsetResponse.schemaVersions()), METADATA(3, "Metadata", MetadataRequestData.SCHEMAS, MetadataResponseData.SCHEMAS), LEADER_AND_ISR(4, "LeaderAndIsr", true, LeaderAndIsrRequestData.SCHEMAS, LeaderAndIsrResponseData.SCHEMAS), diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/RecordsReadable.java b/clients/src/main/java/org/apache/kafka/common/protocol/RecordsReadable.java new file mode 100644 index 0000000000000..5967731a208f5 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/protocol/RecordsReadable.java @@ -0,0 +1,92 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.kafka.common.protocol; + +import org.apache.kafka.common.record.BaseRecords; +import org.apache.kafka.common.record.MemoryRecords; +import org.apache.kafka.common.utils.ByteUtils; + +import java.nio.ByteBuffer; + +/** + * Implementation of Readable which reads from a byte buffer and can read records as {@link MemoryRecords} + * + * @see org.apache.kafka.common.requests.FetchResponse + */ +public class RecordsReadable implements Readable { + private final ByteBuffer buf; + + public RecordsReadable(ByteBuffer buf) { + this.buf = buf; + } + + @Override + public byte readByte() { + return buf.get(); + } + + @Override + public short readShort() { + return buf.getShort(); + } + + @Override + public int readInt() { + return buf.getInt(); + } + + @Override + public long readLong() { + return buf.getLong(); + } + + @Override + public double readDouble() { + return ByteUtils.readDouble(buf); + } + + @Override + public void readArray(byte[] arr) { + buf.get(arr); + } + + @Override + public int readUnsignedVarint() { + return ByteUtils.readUnsignedVarint(buf); + } + + @Override + public ByteBuffer readByteBuffer(int length) { + ByteBuffer res = buf.slice(); + res.limit(length); + + buf.position(buf.position() + length); + + return res; + } + + public BaseRecords readRecords(int length) { + if (length < 0) { + // no records + return null; + } else { + ByteBuffer recordsBuffer = readByteBuffer(length); + return MemoryRecords.readableRecords(recordsBuffer); + } + } +} diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/RecordsWritable.java b/clients/src/main/java/org/apache/kafka/common/protocol/RecordsWritable.java new file mode 100644 index 0000000000000..61f3ee1a452c0 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/protocol/RecordsWritable.java @@ -0,0 +1,139 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.kafka.common.protocol; + +import org.apache.kafka.common.network.ByteBufferSend; +import org.apache.kafka.common.network.Send; +import org.apache.kafka.common.record.BaseRecords; +import org.apache.kafka.common.utils.ByteUtils; + +import java.io.DataOutput; +import java.nio.ByteBuffer; +import java.util.function.Consumer; + +/** + * Implementation of Writable which produces a sequence of {@link Send} objects. This allows for deferring the transfer + * of data from a record-set's file channel to the eventual socket channel. + * + * Excepting {@link #writeRecords(BaseRecords)}, calls to the write methods on this class will append to a byte array + * according to the format specified in {@link DataOutput}. When a call is made to writeRecords, any previously written + * bytes will be flushed as a new {@link ByteBufferSend} to the given Send consumer. After flushing the pending bytes, + * another Send is passed to the consumer which wraps the underlying record-set's transfer logic. + * + * For example, + * + * 
+ *     recordsWritable.writeInt(10);
+ *     recordsWritable.writeRecords(records1);
+ *     recordsWritable.writeInt(20);
+ *     recordsWritable.writeRecords(records2);
+ *     recordsWritable.writeInt(30);
+ *     recordsWritable.flush();
+ *
+ * + * Will pass 5 Send objects to the consumer given in the constructor. Care must be taken by callers to flush any + * pending bytes at the end of the writing sequence to ensure everything is flushed to the consumer. This class is + * intended to be used with {@link org.apache.kafka.common.record.MultiRecordsSend}. + * + * @see org.apache.kafka.common.requests.FetchResponse + */ +public class RecordsWritable implements Writable { + private final String dest; + private final Consumer sendConsumer; + private final ByteBuffer buffer; + private int mark; + + public RecordsWritable(String dest, int messageSizeExcludingRecords, Consumer sendConsumer) { + this.dest = dest; + this.sendConsumer = sendConsumer; + this.buffer = ByteBuffer.allocate(messageSizeExcludingRecords); + this.mark = 0; + } + + @Override + public void writeByte(byte val) { + buffer.put(val); + } + + @Override + public void writeShort(short val) { + buffer.putShort(val); + } + + @Override + public void writeInt(int val) { + buffer.putInt(val); + } + + @Override + public void writeLong(long val) { + buffer.putLong(val); + } + + @Override + public void writeDouble(double val) { + ByteUtils.writeDouble(val, buffer); + } + + @Override + public void writeByteArray(byte[] arr) { + buffer.put(arr); + } + + @Override + public void writeUnsignedVarint(int i) { + ByteUtils.writeUnsignedVarint(i, buffer); + } + + @Override + public void writeByteBuffer(ByteBuffer src) { + buffer.put(src); + } + + public void writeRecords(BaseRecords records) { + flush(); + sendConsumer.accept(records.toSend(dest)); + } + + /** + * Flush any pending bytes as a ByteBufferSend + */ + public void flush() { + int end = buffer.position(); + int len = end - mark; + + if (len > 0) { + int limit = buffer.limit(); + + // Set the desired absolute position and limit before slicing + buffer.position(mark); + buffer.limit(end); + ByteBuffer slice = buffer.slice(); + + // Restore absolute position and limit on original buffer + buffer.limit(limit); + buffer.position(end); + + // Update the mark to the end of slice we just took + mark = end; + + ByteBufferSend send = new ByteBufferSend(dest, slice); + sendConsumer.accept(send); + } + } +} diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java index f4085b833b21c..b8266d06c9a9f 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java @@ -17,6 +17,7 @@ package org.apache.kafka.common.requests; import org.apache.kafka.common.errors.UnsupportedVersionException; +import org.apache.kafka.common.message.FetchRequestData; import org.apache.kafka.common.network.NetworkSend; import org.apache.kafka.common.network.Send; import org.apache.kafka.common.protocol.ApiKeys; @@ -146,7 +147,7 @@ public static AbstractRequest parseRequest(ApiKeys apiKey, short apiVersion, Str case PRODUCE: return new ProduceRequest(struct, apiVersion); case FETCH: - return new FetchRequest(struct, apiVersion); + return new FetchRequest(new FetchRequestData(struct, apiVersion), apiVersion); case LIST_OFFSETS: return new ListOffsetRequest(struct, apiVersion); case METADATA: diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequestResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequestResponse.java index b02659d4f354c..39698a1d20b74 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequestResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequestResponse.java @@ -17,4 +17,5 @@ package org.apache.kafka.common.requests; public interface AbstractRequestResponse { + } diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java index 32581421bb940..ed21a0ebfc80f 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.common.requests; +import org.apache.kafka.common.message.FetchResponseData; import org.apache.kafka.common.network.NetworkSend; import org.apache.kafka.common.network.Send; import org.apache.kafka.common.protocol.ApiKeys; @@ -89,7 +90,7 @@ public static AbstractResponse parseResponse(ApiKeys apiKey, Struct struct, shor case PRODUCE: return new ProduceResponse(struct); case FETCH: - return FetchResponse.parse(struct); + return new FetchResponse<>(new FetchResponseData(struct, version)); case LIST_OFFSETS: return new ListOffsetResponse(struct); case METADATA: diff --git a/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java index 29cbb6010b5be..01b5e1a7b5b3c 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java @@ -18,221 +18,46 @@ import org.apache.kafka.common.IsolationLevel; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.message.FetchRequestData; +import org.apache.kafka.common.message.RequestHeaderData; import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.ByteBufferAccessor; import org.apache.kafka.common.protocol.Errors; -import org.apache.kafka.common.protocol.types.Field; -import org.apache.kafka.common.protocol.types.Schema; +import org.apache.kafka.common.protocol.ObjectSerializationCache; import org.apache.kafka.common.protocol.types.Struct; -import org.apache.kafka.common.protocol.types.Type; import org.apache.kafka.common.record.MemoryRecords; +import org.apache.kafka.common.record.RecordBatch; import org.apache.kafka.common.utils.Utils; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collections; -import java.util.HashMap; import java.util.Iterator; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Optional; - -import static org.apache.kafka.common.protocol.CommonFields.CURRENT_LEADER_EPOCH; -import static org.apache.kafka.common.protocol.CommonFields.PARTITION_ID; -import static org.apache.kafka.common.protocol.CommonFields.TOPIC_NAME; -import static org.apache.kafka.common.requests.FetchMetadata.FINAL_EPOCH; -import static org.apache.kafka.common.requests.FetchMetadata.INVALID_SESSION_ID; +import java.util.stream.Collectors; public class FetchRequest extends AbstractRequest { - public static final int CONSUMER_REPLICA_ID = -1; - private static final Field.ComplexArray TOPICS = new Field.ComplexArray("topics", - "Topics to fetch in the order provided."); - private static final Field.ComplexArray FORGOTTEN_TOPICS = new Field.ComplexArray("forgotten_topics_data", - "Topics to remove from the fetch session."); - private static final Field.Int32 MAX_BYTES = new Field.Int32("max_bytes", - "Maximum bytes to accumulate in the response. Note that this is not an absolute maximum, " + - "if the first message in the first non-empty partition of the fetch is larger than this " + - "value, the message will still be returned to ensure that progress can be made."); - private static final Field.Int8 ISOLATION_LEVEL = new Field.Int8("isolation_level", - "This setting controls the visibility of transactional records. Using READ_UNCOMMITTED " + - "(isolation_level = 0) makes all records visible. With READ_COMMITTED (isolation_level = 1), " + - "non-transactional and COMMITTED transactional records are visible. To be more concrete, " + - "READ_COMMITTED returns all data from offsets smaller than the current LSO (last stable offset), " + - "and enables the inclusion of the list of aborted transactions in the result, which allows " + - "consumers to discard ABORTED transactional records"); - private static final Field.Int32 SESSION_ID = new Field.Int32("session_id", "The fetch session ID"); - private static final Field.Int32 SESSION_EPOCH = new Field.Int32("session_epoch", "The fetch session epoch"); - private static final Field.Str RACK_ID = new Field.Str("rack_id", "The consumer's rack id"); - - // topic level fields - private static final Field.ComplexArray PARTITIONS = new Field.ComplexArray("partitions", - "Partitions to fetch."); - - // partition level fields - private static final Field.Int32 REPLICA_ID = new Field.Int32("replica_id", - "Broker id of the follower. For normal consumers, use -1."); - private static final Field.Int64 FETCH_OFFSET = new Field.Int64("fetch_offset", "Message offset."); - private static final Field.Int32 PARTITION_MAX_BYTES = new Field.Int32("partition_max_bytes", - "Maximum bytes to fetch."); - private static final Field.Int32 MAX_WAIT_TIME = new Field.Int32("max_wait_time", - "Maximum time in ms to wait for the response."); - private static final Field.Int32 MIN_BYTES = new Field.Int32("min_bytes", - "Minimum bytes to accumulate in the response."); - private static final Field.Int64 LOG_START_OFFSET = new Field.Int64("log_start_offset", - "Earliest available offset of the follower replica. " + - "The field is only used when request is sent by follower. "); - - private static final Field PARTITIONS_V0 = PARTITIONS.withFields( - PARTITION_ID, - FETCH_OFFSET, - PARTITION_MAX_BYTES); - - private static final Field TOPICS_V0 = TOPICS.withFields( - TOPIC_NAME, - PARTITIONS_V0); - - private static final Schema FETCH_REQUEST_V0 = new Schema( - REPLICA_ID, - MAX_WAIT_TIME, - MIN_BYTES, - TOPICS_V0); - - // The V1 Fetch Request body is the same as V0. - // Only the version number is incremented to indicate a newer client - private static final Schema FETCH_REQUEST_V1 = FETCH_REQUEST_V0; - - // V2 bumped to indicate the client support message format V1 which uses relative offset and has timestamp. - private static final Schema FETCH_REQUEST_V2 = FETCH_REQUEST_V1; - - // V3 added top level max_bytes field - the total size of partition data to accumulate in response. - // The partition ordering is now relevant - partitions will be processed in order they appear in request. - private static final Schema FETCH_REQUEST_V3 = new Schema( - REPLICA_ID, - MAX_WAIT_TIME, - MIN_BYTES, - MAX_BYTES, - TOPICS_V0); - - // V4 adds the fetch isolation level and exposes magic v2 (via the response). - - private static final Schema FETCH_REQUEST_V4 = new Schema( - REPLICA_ID, - MAX_WAIT_TIME, - MIN_BYTES, - MAX_BYTES, - ISOLATION_LEVEL, - TOPICS_V0); - - - // V5 added log_start_offset field - the earliest available offset of partition data that can be consumed. - private static final Field PARTITIONS_V5 = PARTITIONS.withFields( - PARTITION_ID, - FETCH_OFFSET, - LOG_START_OFFSET, - PARTITION_MAX_BYTES); - - private static final Field TOPICS_V5 = TOPICS.withFields( - TOPIC_NAME, - PARTITIONS_V5); - - private static final Schema FETCH_REQUEST_V5 = new Schema( - REPLICA_ID, - MAX_WAIT_TIME, - MIN_BYTES, - MAX_BYTES, - ISOLATION_LEVEL, - TOPICS_V5); - - // V6 bumped up to indicate that the client supports KafkaStorageException. The KafkaStorageException will be - // translated to NotLeaderOrFollowerException in the response if version <= 5 - private static final Schema FETCH_REQUEST_V6 = FETCH_REQUEST_V5; - - // V7 added incremental fetch requests. - private static final Field.Array FORGOTTEN_PARTITIONS = new Field.Array("partitions", Type.INT32, - "Partitions to remove from the fetch session."); - private static final Field FORGOTTEN_TOPIC_DATA_V7 = FORGOTTEN_TOPICS.withFields( - TOPIC_NAME, - FORGOTTEN_PARTITIONS); - - private static final Schema FETCH_REQUEST_V7 = new Schema( - REPLICA_ID, - MAX_WAIT_TIME, - MIN_BYTES, - MAX_BYTES, - ISOLATION_LEVEL, - SESSION_ID, - SESSION_EPOCH, - TOPICS_V5, - FORGOTTEN_TOPIC_DATA_V7); - - // V8 bump used to indicate that on quota violation brokers send out responses before throttling. - private static final Schema FETCH_REQUEST_V8 = FETCH_REQUEST_V7; - - // V9 adds the current leader epoch (see KIP-320) - private static final Field FETCH_REQUEST_PARTITION_V9 = PARTITIONS.withFields( - PARTITION_ID, - CURRENT_LEADER_EPOCH, - FETCH_OFFSET, - LOG_START_OFFSET, - PARTITION_MAX_BYTES); - - private static final Field FETCH_REQUEST_TOPIC_V9 = TOPICS.withFields( - TOPIC_NAME, - FETCH_REQUEST_PARTITION_V9); - - private static final Schema FETCH_REQUEST_V9 = new Schema( - REPLICA_ID, - MAX_WAIT_TIME, - MIN_BYTES, - MAX_BYTES, - ISOLATION_LEVEL, - SESSION_ID, - SESSION_EPOCH, - FETCH_REQUEST_TOPIC_V9, - FORGOTTEN_TOPIC_DATA_V7); - - // V10 bumped up to indicate ZStandard capability. (see KIP-110) - private static final Schema FETCH_REQUEST_V10 = FETCH_REQUEST_V9; - - // V11 added rack ID to support read from followers (KIP-392) - private static final Schema FETCH_REQUEST_V11 = new Schema( - REPLICA_ID, - MAX_WAIT_TIME, - MIN_BYTES, - MAX_BYTES, - ISOLATION_LEVEL, - SESSION_ID, - SESSION_EPOCH, - FETCH_REQUEST_TOPIC_V9, - FORGOTTEN_TOPIC_DATA_V7, - RACK_ID); - - public static Schema[] schemaVersions() { - return new Schema[]{FETCH_REQUEST_V0, FETCH_REQUEST_V1, FETCH_REQUEST_V2, FETCH_REQUEST_V3, FETCH_REQUEST_V4, - FETCH_REQUEST_V5, FETCH_REQUEST_V6, FETCH_REQUEST_V7, FETCH_REQUEST_V8, FETCH_REQUEST_V9, - FETCH_REQUEST_V10, FETCH_REQUEST_V11}; - } + public static final int CONSUMER_REPLICA_ID = -1; // default values for older versions where a request level limit did not exist public static final int DEFAULT_RESPONSE_MAX_BYTES = Integer.MAX_VALUE; public static final long INVALID_LOG_START_OFFSET = -1L; - private final int replicaId; - private final int maxWait; - private final int minBytes; - private final int maxBytes; - private final IsolationLevel isolationLevel; + private final FetchRequestData data; - // Note: the iteration order of this map is significant, since it determines the order - // in which partitions appear in the message. For this reason, this map should have a - // deterministic iteration order, like LinkedHashMap or TreeMap (but unlike HashMap). + // These are immutable read-only structures derived from FetchRequestData private final Map fetchData; - private final List toForget; private final FetchMetadata metadata; - private final String rackId; + + public FetchRequestData data() { + return data; + } public static final class PartitionData { public final long fetchOffset; @@ -273,6 +98,28 @@ public boolean equals(Object o) { } } + private Map toPartitionDataMap(List fetchableTopics) { + Map result = new LinkedHashMap<>(); + fetchableTopics.forEach(fetchTopic -> fetchTopic.partitions().forEach(fetchPartition -> { + Optional leaderEpoch = Optional.of(fetchPartition.currentLeaderEpoch()) + .filter(epoch -> epoch != RecordBatch.NO_PARTITION_LEADER_EPOCH); + result.put(new TopicPartition(fetchTopic.topic(), fetchPartition.partition()), + new PartitionData(fetchPartition.fetchOffset(), fetchPartition.logStartOffset(), + fetchPartition.partitionMaxBytes(), leaderEpoch)); + })); + return Collections.unmodifiableMap(result); + } + + private List toForgottenTopicList(List forgottenTopics) { + List result = new ArrayList<>(); + forgottenTopics.forEach(forgottenTopic -> + forgottenTopic.partitions().forEach(partitionId -> + result.add(new TopicPartition(forgottenTopic.topic(), partitionId)) + ) + ); + return result; + } + static final class TopicAndPartitionData { public final String topic; public final LinkedHashMap partitions; @@ -366,8 +213,47 @@ public FetchRequest build(short version) { maxBytes = DEFAULT_RESPONSE_MAX_BYTES; } - return new FetchRequest(version, replicaId, maxWait, minBytes, maxBytes, fetchData, - isolationLevel, toForget, metadata, rackId); + FetchRequestData fetchRequestData = new FetchRequestData(); + fetchRequestData.setReplicaId(replicaId); + fetchRequestData.setMaxWaitMs(maxWait); + fetchRequestData.setMinBytes(minBytes); + fetchRequestData.setMaxBytes(maxBytes); + fetchRequestData.setIsolationLevel(isolationLevel.id()); + fetchRequestData.setForgottenTopicsData(new ArrayList<>()); + toForget.stream() + .collect(Collectors.groupingBy(TopicPartition::topic, LinkedHashMap::new, Collectors.toList())) + .forEach((topic, partitions) -> + fetchRequestData.forgottenTopicsData().add(new FetchRequestData.ForgottenTopic() + .setTopic(topic) + .setPartitions(partitions.stream().map(TopicPartition::partition).collect(Collectors.toList()))) + ); + fetchRequestData.setTopics(new ArrayList<>()); + + // We collect the partitions in a single FetchTopic only if they appear sequentially in the fetchData + FetchRequestData.FetchTopic fetchTopic = null; + for (Map.Entry entry : fetchData.entrySet()) { + if (fetchTopic == null || !entry.getKey().topic().equals(fetchTopic.topic())) { + fetchTopic = new FetchRequestData.FetchTopic() + .setTopic(entry.getKey().topic()) + .setPartitions(new ArrayList<>()); + fetchRequestData.topics().add(fetchTopic); + } + + fetchTopic.partitions().add( + new FetchRequestData.FetchPartition().setPartition(entry.getKey().partition()) + .setCurrentLeaderEpoch(entry.getValue().currentLeaderEpoch.orElse(RecordBatch.NO_PARTITION_LEADER_EPOCH)) + .setFetchOffset(entry.getValue().fetchOffset) + .setLogStartOffset(entry.getValue().logStartOffset) + .setPartitionMaxBytes(entry.getValue().maxBytes)); + } + + if (metadata != null) { + fetchRequestData.setSessionEpoch(metadata.epoch()); + fetchRequestData.setSessionId(metadata.sessionId()); + } + fetchRequestData.setRackId(rackId); + + return new FetchRequest(fetchRequestData, version); } @Override @@ -388,64 +274,12 @@ public String toString() { } } - private FetchRequest(short version, int replicaId, int maxWait, int minBytes, int maxBytes, - Map fetchData, IsolationLevel isolationLevel, - List toForget, FetchMetadata metadata, String rackId) { - super(ApiKeys.FETCH, version); - this.replicaId = replicaId; - this.maxWait = maxWait; - this.minBytes = minBytes; - this.maxBytes = maxBytes; - this.fetchData = fetchData; - this.isolationLevel = isolationLevel; - this.toForget = toForget; - this.metadata = metadata; - this.rackId = rackId; - } - - public FetchRequest(Struct struct, short version) { + public FetchRequest(FetchRequestData fetchRequestData, short version) { super(ApiKeys.FETCH, version); - replicaId = struct.get(REPLICA_ID); - maxWait = struct.get(MAX_WAIT_TIME); - minBytes = struct.get(MIN_BYTES); - maxBytes = struct.getOrElse(MAX_BYTES, DEFAULT_RESPONSE_MAX_BYTES); - - if (struct.hasField(ISOLATION_LEVEL)) - isolationLevel = IsolationLevel.forId(struct.get(ISOLATION_LEVEL)); - else - isolationLevel = IsolationLevel.READ_UNCOMMITTED; - toForget = new ArrayList<>(0); - if (struct.hasField(FORGOTTEN_TOPICS)) { - for (Object forgottenTopicObj : struct.get(FORGOTTEN_TOPICS)) { - Struct forgottenTopic = (Struct) forgottenTopicObj; - String topicName = forgottenTopic.get(TOPIC_NAME); - for (Object partObj : forgottenTopic.get(FORGOTTEN_PARTITIONS)) { - Integer part = (Integer) partObj; - toForget.add(new TopicPartition(topicName, part)); - } - } - } - metadata = new FetchMetadata(struct.getOrElse(SESSION_ID, INVALID_SESSION_ID), - struct.getOrElse(SESSION_EPOCH, FINAL_EPOCH)); - - fetchData = new LinkedHashMap<>(); - for (Object topicResponseObj : struct.get(TOPICS)) { - Struct topicResponse = (Struct) topicResponseObj; - String topic = topicResponse.get(TOPIC_NAME); - for (Object partitionResponseObj : topicResponse.get(PARTITIONS)) { - Struct partitionResponse = (Struct) partitionResponseObj; - int partition = partitionResponse.get(PARTITION_ID); - long offset = partitionResponse.get(FETCH_OFFSET); - int maxBytes = partitionResponse.get(PARTITION_MAX_BYTES); - long logStartOffset = partitionResponse.getOrElse(LOG_START_OFFSET, INVALID_LOG_START_OFFSET); - - // Current leader epoch added in v9 - Optional currentLeaderEpoch = RequestUtils.getLeaderEpoch(partitionResponse, CURRENT_LEADER_EPOCH); - PartitionData partitionData = new PartitionData(offset, logStartOffset, maxBytes, currentLeaderEpoch); - fetchData.put(new TopicPartition(topic, partition), partitionData); - } - } - rackId = struct.getOrElse(RACK_ID, ""); + this.data = fetchRequestData; + this.fetchData = toPartitionDataMap(fetchRequestData.topics()); + this.toForget = toForgottenTopicList(fetchRequestData.forgottenTopicsData()); + this.metadata = new FetchMetadata(fetchRequestData.sessionId(), fetchRequestData.sessionEpoch()); } @Override @@ -464,23 +298,23 @@ public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) { FetchResponse.INVALID_LOG_START_OFFSET, Optional.empty(), null, MemoryRecords.EMPTY); responseData.put(entry.getKey(), partitionResponse); } - return new FetchResponse<>(error, responseData, throttleTimeMs, metadata.sessionId()); + return new FetchResponse<>(error, responseData, throttleTimeMs, data.sessionId()); } public int replicaId() { - return replicaId; + return data.replicaId(); } public int maxWait() { - return maxWait; + return data.maxWaitMs(); } public int minBytes() { - return minBytes; + return data.minBytes(); } public int maxBytes() { - return maxBytes; + return data.maxBytes(); } public Map fetchData() { @@ -492,11 +326,11 @@ public List toForget() { } public boolean isFromFollower() { - return replicaId >= 0; + return replicaId() >= 0; } public IsolationLevel isolationLevel() { - return isolationLevel; + return IsolationLevel.forId(data.isolationLevel()); } public FetchMetadata metadata() { @@ -504,62 +338,37 @@ public FetchMetadata metadata() { } public String rackId() { - return rackId; + return data.rackId(); } + @Override + public ByteBuffer serialize(RequestHeader header) { + // Unlike the custom FetchResponse#toSend, we don't include the buffer size here. This buffer is passed + // to a NetworkSend which adds the length value in the eventual serialization + + ObjectSerializationCache cache = new ObjectSerializationCache(); + RequestHeaderData requestHeaderData = header.data(); + + int headerSize = requestHeaderData.size(cache, header.headerVersion()); + int bodySize = data.size(cache, header.apiVersion()); + + ByteBuffer buffer = ByteBuffer.allocate(headerSize + bodySize); + ByteBufferAccessor writer = new ByteBufferAccessor(buffer); + + requestHeaderData.write(writer, cache, header.headerVersion()); + data.write(writer, cache, header.apiVersion()); + + buffer.rewind(); + return buffer; + } + + // For testing public static FetchRequest parse(ByteBuffer buffer, short version) { - return new FetchRequest(ApiKeys.FETCH.parseRequest(version, buffer), version); + return new FetchRequest(new FetchRequestData(ApiKeys.FETCH.parseRequest(version, buffer), version), version); } @Override protected Struct toStruct() { - Struct struct = new Struct(ApiKeys.FETCH.requestSchema(version())); - List> topicsData = - TopicAndPartitionData.batchByTopic(fetchData.entrySet().iterator()); - - struct.set(REPLICA_ID, replicaId); - struct.set(MAX_WAIT_TIME, maxWait); - struct.set(MIN_BYTES, minBytes); - struct.setIfExists(MAX_BYTES, maxBytes); - struct.setIfExists(ISOLATION_LEVEL, isolationLevel.id()); - struct.setIfExists(SESSION_ID, metadata.sessionId()); - struct.setIfExists(SESSION_EPOCH, metadata.epoch()); - - List topicArray = new ArrayList<>(); - for (TopicAndPartitionData topicEntry : topicsData) { - Struct topicData = struct.instance(TOPICS); - topicData.set(TOPIC_NAME, topicEntry.topic); - List partitionArray = new ArrayList<>(); - for (Map.Entry partitionEntry : topicEntry.partitions.entrySet()) { - PartitionData fetchPartitionData = partitionEntry.getValue(); - Struct partitionData = topicData.instance(PARTITIONS); - partitionData.set(PARTITION_ID, partitionEntry.getKey()); - partitionData.set(FETCH_OFFSET, fetchPartitionData.fetchOffset); - partitionData.set(PARTITION_MAX_BYTES, fetchPartitionData.maxBytes); - partitionData.setIfExists(LOG_START_OFFSET, fetchPartitionData.logStartOffset); - RequestUtils.setLeaderEpochIfExists(partitionData, CURRENT_LEADER_EPOCH, fetchPartitionData.currentLeaderEpoch); - partitionArray.add(partitionData); - } - topicData.set(PARTITIONS, partitionArray.toArray()); - topicArray.add(topicData); - } - struct.set(TOPICS, topicArray.toArray()); - if (struct.hasField(FORGOTTEN_TOPICS)) { - Map> topicsToPartitions = new HashMap<>(); - for (TopicPartition part : toForget) { - List partitions = topicsToPartitions.computeIfAbsent(part.topic(), topic -> new ArrayList<>()); - partitions.add(part.partition()); - } - List toForgetStructs = new ArrayList<>(); - for (Map.Entry> entry : topicsToPartitions.entrySet()) { - Struct toForgetStruct = struct.instance(FORGOTTEN_TOPICS); - toForgetStruct.set(TOPIC_NAME, entry.getKey()); - toForgetStruct.set(FORGOTTEN_PARTITIONS, entry.getValue().toArray()); - toForgetStructs.add(toForgetStruct); - } - struct.set(FORGOTTEN_TOPICS, toForgetStructs.toArray()); - } - struct.setIfExists(RACK_ID, rackId); - return struct; + return data.toStruct(version()); } } diff --git a/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java index 984c50ca724bd..cb9a5279e84ee 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java @@ -17,18 +17,19 @@ package org.apache.kafka.common.requests; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.message.FetchResponseData; +import org.apache.kafka.common.message.ResponseHeaderData; import org.apache.kafka.common.network.ByteBufferSend; import org.apache.kafka.common.network.Send; -import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.ByteBufferAccessor; import org.apache.kafka.common.protocol.Errors; -import org.apache.kafka.common.protocol.types.ArrayOf; -import org.apache.kafka.common.protocol.types.Field; -import org.apache.kafka.common.protocol.types.Schema; +import org.apache.kafka.common.protocol.ObjectSerializationCache; +import org.apache.kafka.common.protocol.RecordsReadable; +import org.apache.kafka.common.protocol.RecordsWritable; import org.apache.kafka.common.protocol.types.Struct; import org.apache.kafka.common.record.BaseRecords; import org.apache.kafka.common.record.MemoryRecords; import org.apache.kafka.common.record.MultiRecordsSend; -import org.apache.kafka.common.record.RecordsSend; import java.nio.ByteBuffer; import java.util.ArrayDeque; @@ -38,17 +39,9 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.Objects; import java.util.Optional; -import java.util.Queue; -import java.util.function.Predicate; - -import static org.apache.kafka.common.protocol.CommonFields.ERROR_CODE; -import static org.apache.kafka.common.protocol.CommonFields.PARTITION_ID; -import static org.apache.kafka.common.protocol.CommonFields.THROTTLE_TIME_MS; -import static org.apache.kafka.common.protocol.CommonFields.TOPIC_NAME; -import static org.apache.kafka.common.protocol.types.Type.RECORDS; -import static org.apache.kafka.common.protocol.types.Type.STRING; +import java.util.stream.Collectors; + import static org.apache.kafka.common.requests.FetchMetadata.INVALID_SESSION_ID; /** @@ -73,168 +66,18 @@ */ public class FetchResponse extends AbstractResponse { - private static final String RESPONSES_KEY_NAME = "responses"; - - // topic level field names - private static final String PARTITIONS_KEY_NAME = "partition_responses"; - - // partition level fields - private static final Field.Int64 HIGH_WATERMARK = new Field.Int64("high_watermark", - "Last committed offset."); - private static final Field.Int64 LOG_START_OFFSET = new Field.Int64("log_start_offset", - "Earliest available offset."); - private static final Field.Int32 PREFERRED_READ_REPLICA = new Field.Int32("preferred_read_replica", - "The ID of the replica that the consumer should prefer."); - - private static final String PARTITION_HEADER_KEY_NAME = "partition_header"; - private static final String ABORTED_TRANSACTIONS_KEY_NAME = "aborted_transactions"; - private static final String RECORD_SET_KEY_NAME = "record_set"; - - private static final Schema FETCH_RESPONSE_PARTITION_HEADER_V0 = new Schema( - PARTITION_ID, - ERROR_CODE, - HIGH_WATERMARK); - private static final Schema FETCH_RESPONSE_PARTITION_V0 = new Schema( - new Field(PARTITION_HEADER_KEY_NAME, FETCH_RESPONSE_PARTITION_HEADER_V0), - new Field(RECORD_SET_KEY_NAME, RECORDS)); - - private static final Schema FETCH_RESPONSE_TOPIC_V0 = new Schema( - TOPIC_NAME, - new Field(PARTITIONS_KEY_NAME, new ArrayOf(FETCH_RESPONSE_PARTITION_V0))); - - private static final Schema FETCH_RESPONSE_V0 = new Schema( - new Field(RESPONSES_KEY_NAME, new ArrayOf(FETCH_RESPONSE_TOPIC_V0))); - - // V1 bumped for the addition of the throttle time - private static final Schema FETCH_RESPONSE_V1 = new Schema( - THROTTLE_TIME_MS, - new Field(RESPONSES_KEY_NAME, new ArrayOf(FETCH_RESPONSE_TOPIC_V0))); - - // V2 bumped to indicate the client support message format V1 which uses relative offset and has timestamp. - private static final Schema FETCH_RESPONSE_V2 = FETCH_RESPONSE_V1; - - // V3 bumped for addition of top-levl max_bytes field and to indicate that partition ordering is relevant - private static final Schema FETCH_RESPONSE_V3 = FETCH_RESPONSE_V2; - - // V4 adds features for transactional consumption (the aborted transaction list and the - // last stable offset). It also exposes messages with magic v2 (along with older formats). - // aborted transaction field names - private static final Field.Int64 LAST_STABLE_OFFSET = new Field.Int64("last_stable_offset", - "The last stable offset (or LSO) of the partition. This is the last offset such that the state " + - "of all transactional records prior to this offset have been decided (ABORTED or COMMITTED)"); - private static final Field.Int64 PRODUCER_ID = new Field.Int64("producer_id", - "The producer id associated with the aborted transactions"); - private static final Field.Int64 FIRST_OFFSET = new Field.Int64("first_offset", - "The first offset in the aborted transaction"); - - private static final Schema FETCH_RESPONSE_ABORTED_TRANSACTION_V4 = new Schema( - PRODUCER_ID, - FIRST_OFFSET); - - private static final Schema FETCH_RESPONSE_PARTITION_HEADER_V4 = new Schema( - PARTITION_ID, - ERROR_CODE, - HIGH_WATERMARK, - LAST_STABLE_OFFSET, - new Field(ABORTED_TRANSACTIONS_KEY_NAME, ArrayOf.nullable(FETCH_RESPONSE_ABORTED_TRANSACTION_V4))); - - // V5 added log_start_offset field - the earliest available offset of partition data that can be consumed. - private static final Schema FETCH_RESPONSE_PARTITION_HEADER_V5 = new Schema( - PARTITION_ID, - ERROR_CODE, - HIGH_WATERMARK, - LAST_STABLE_OFFSET, - LOG_START_OFFSET, - new Field(ABORTED_TRANSACTIONS_KEY_NAME, ArrayOf.nullable(FETCH_RESPONSE_ABORTED_TRANSACTION_V4))); - - // Introduced in V11 to support read from followers (KIP-392) - private static final Schema FETCH_RESPONSE_PARTITION_HEADER_V6 = new Schema( - PARTITION_ID, - ERROR_CODE, - HIGH_WATERMARK, - LAST_STABLE_OFFSET, - LOG_START_OFFSET, - new Field(ABORTED_TRANSACTIONS_KEY_NAME, ArrayOf.nullable(FETCH_RESPONSE_ABORTED_TRANSACTION_V4)), - PREFERRED_READ_REPLICA); - - private static final Schema FETCH_RESPONSE_PARTITION_V4 = new Schema( - new Field(PARTITION_HEADER_KEY_NAME, FETCH_RESPONSE_PARTITION_HEADER_V4), - new Field(RECORD_SET_KEY_NAME, RECORDS)); - - private static final Schema FETCH_RESPONSE_PARTITION_V5 = new Schema( - new Field(PARTITION_HEADER_KEY_NAME, FETCH_RESPONSE_PARTITION_HEADER_V5), - new Field(RECORD_SET_KEY_NAME, RECORDS)); - - private static final Schema FETCH_RESPONSE_PARTITION_V6 = new Schema( - new Field(PARTITION_HEADER_KEY_NAME, FETCH_RESPONSE_PARTITION_HEADER_V6), - new Field(RECORD_SET_KEY_NAME, RECORDS)); - - private static final Schema FETCH_RESPONSE_TOPIC_V4 = new Schema( - TOPIC_NAME, - new Field(PARTITIONS_KEY_NAME, new ArrayOf(FETCH_RESPONSE_PARTITION_V4))); - - private static final Schema FETCH_RESPONSE_TOPIC_V5 = new Schema( - TOPIC_NAME, - new Field(PARTITIONS_KEY_NAME, new ArrayOf(FETCH_RESPONSE_PARTITION_V5))); - - private static final Schema FETCH_RESPONSE_TOPIC_V6 = new Schema( - TOPIC_NAME, - new Field(PARTITIONS_KEY_NAME, new ArrayOf(FETCH_RESPONSE_PARTITION_V6))); - - private static final Schema FETCH_RESPONSE_V4 = new Schema( - THROTTLE_TIME_MS, - new Field(RESPONSES_KEY_NAME, new ArrayOf(FETCH_RESPONSE_TOPIC_V4))); - - private static final Schema FETCH_RESPONSE_V5 = new Schema( - THROTTLE_TIME_MS, - new Field(RESPONSES_KEY_NAME, new ArrayOf(FETCH_RESPONSE_TOPIC_V5))); - - // V6 bumped up to indicate that the client supports KafkaStorageException. The KafkaStorageException will - // be translated to NotLeaderOrFollowerException in the response if version <= 5 - private static final Schema FETCH_RESPONSE_V6 = FETCH_RESPONSE_V5; - - // V7 added incremental fetch responses and a top-level error code. - private static final Field.Int32 SESSION_ID = new Field.Int32("session_id", "The fetch session ID"); - - private static final Schema FETCH_RESPONSE_V7 = new Schema( - THROTTLE_TIME_MS, - ERROR_CODE, - SESSION_ID, - new Field(RESPONSES_KEY_NAME, new ArrayOf(FETCH_RESPONSE_TOPIC_V5))); - - // V8 bump used to indicate that on quota violation brokers send out responses before throttling. - private static final Schema FETCH_RESPONSE_V8 = FETCH_RESPONSE_V7; - - // V9 adds the current leader epoch (see KIP-320) - private static final Schema FETCH_RESPONSE_V9 = FETCH_RESPONSE_V8; - - // V10 bumped up to indicate ZStandard capability. (see KIP-110) - private static final Schema FETCH_RESPONSE_V10 = FETCH_RESPONSE_V9; - - // V11 added preferred read replica for each partition response to support read from followers (KIP-392) - private static final Schema FETCH_RESPONSE_V11 = new Schema( - THROTTLE_TIME_MS, - ERROR_CODE, - SESSION_ID, - new Field(RESPONSES_KEY_NAME, new ArrayOf(FETCH_RESPONSE_TOPIC_V6))); - - - public static Schema[] schemaVersions() { - return new Schema[] {FETCH_RESPONSE_V0, FETCH_RESPONSE_V1, FETCH_RESPONSE_V2, - FETCH_RESPONSE_V3, FETCH_RESPONSE_V4, FETCH_RESPONSE_V5, FETCH_RESPONSE_V6, - FETCH_RESPONSE_V7, FETCH_RESPONSE_V8, FETCH_RESPONSE_V9, FETCH_RESPONSE_V10, - FETCH_RESPONSE_V11}; - } - public static final long INVALID_HIGHWATERMARK = -1L; public static final long INVALID_LAST_STABLE_OFFSET = -1L; public static final long INVALID_LOG_START_OFFSET = -1L; public static final int INVALID_PREFERRED_REPLICA_ID = -1; - private final int throttleTimeMs; - private final Errors error; - private final int sessionId; - private final LinkedHashMap> responseData; + private final FetchResponseData data; + private final LinkedHashMap> responseDataMap; + + public FetchResponseData data() { + return data; + } + public static final class AbortedTransaction { public final long producerId; @@ -268,16 +111,38 @@ public int hashCode() { public String toString() { return "(producerId=" + producerId + ", firstOffset=" + firstOffset + ")"; } + + static AbortedTransaction fromMessage(FetchResponseData.AbortedTransaction abortedTransaction) { + return new AbortedTransaction(abortedTransaction.producerId(), abortedTransaction.firstOffset()); + } } public static final class PartitionData { - public final Errors error; - public final long highWatermark; - public final long lastStableOffset; - public final long logStartOffset; - public final Optional preferredReadReplica; - public final List abortedTransactions; - public final T records; + private final FetchResponseData.FetchablePartitionResponse partitionResponse; + + // Derived fields + private final Optional preferredReplica; + private final List abortedTransactions; + private final Errors error; + + private PartitionData(FetchResponseData.FetchablePartitionResponse partitionResponse) { + // We partially construct FetchablePartitionResponse since we don't know the partition ID at this point + // When we convert the PartitionData (and other fields) into FetchResponseData down in toMessage, we + // set the partition IDs. + this.partitionResponse = partitionResponse; + this.preferredReplica = Optional.of(partitionResponse.partitionHeader().preferredReadReplica()) + .filter(replicaId -> replicaId != INVALID_PREFERRED_REPLICA_ID); + + if (partitionResponse.partitionHeader().abortedTransactions() == null) { + this.abortedTransactions = null; + } else { + this.abortedTransactions = partitionResponse.partitionHeader().abortedTransactions().stream() + .map(AbortedTransaction::fromMessage) + .collect(Collectors.toList()); + } + + this.error = Errors.forCode(partitionResponse.partitionHeader().errorCode()); + } public PartitionData(Errors error, long highWatermark, @@ -286,13 +151,27 @@ public PartitionData(Errors error, Optional preferredReadReplica, List abortedTransactions, T records) { - this.error = error; - this.highWatermark = highWatermark; - this.lastStableOffset = lastStableOffset; - this.logStartOffset = logStartOffset; - this.preferredReadReplica = preferredReadReplica; + this.preferredReplica = preferredReadReplica; this.abortedTransactions = abortedTransactions; - this.records = records; + this.error = error; + FetchResponseData.PartitionHeader partitionHeader = new FetchResponseData.PartitionHeader(); + partitionHeader.setErrorCode(error.code()) + .setHighWatermark(highWatermark) + .setLastStableOffset(lastStableOffset) + .setLogStartOffset(logStartOffset); + if (abortedTransactions != null) { + partitionHeader.setAbortedTransactions(abortedTransactions.stream().map( + aborted -> new FetchResponseData.AbortedTransaction() + .setProducerId(aborted.producerId) + .setFirstOffset(aborted.firstOffset)) + .collect(Collectors.toList())); + } else { + partitionHeader.setAbortedTransactions(null); + } + partitionHeader.setPreferredReadReplica(preferredReadReplica.orElse(INVALID_PREFERRED_REPLICA_ID)); + this.partitionResponse = new FetchResponseData.FetchablePartitionResponse() + .setPartitionHeader(partitionHeader) + .setRecordSet(records); } public PartitionData(Errors error, @@ -301,13 +180,7 @@ public PartitionData(Errors error, long logStartOffset, List abortedTransactions, T records) { - this.error = error; - this.highWatermark = highWatermark; - this.lastStableOffset = lastStableOffset; - this.logStartOffset = logStartOffset; - this.preferredReadReplica = Optional.empty(); - this.abortedTransactions = abortedTransactions; - this.records = records; + this(error, highWatermark, lastStableOffset, logStartOffset, Optional.empty(), abortedTransactions, records); } @Override @@ -319,38 +192,53 @@ public boolean equals(Object o) { PartitionData that = (PartitionData) o; - return error == that.error && - highWatermark == that.highWatermark && - lastStableOffset == that.lastStableOffset && - logStartOffset == that.logStartOffset && - Objects.equals(preferredReadReplica, that.preferredReadReplica) && - Objects.equals(abortedTransactions, that.abortedTransactions) && - Objects.equals(records, that.records); + return this.partitionResponse.equals(that.partitionResponse); } @Override public int hashCode() { - int result = error != null ? error.hashCode() : 0; - result = 31 * result + Long.hashCode(highWatermark); - result = 31 * result + Long.hashCode(lastStableOffset); - result = 31 * result + Long.hashCode(logStartOffset); - result = 31 * result + Objects.hashCode(preferredReadReplica); - result = 31 * result + (abortedTransactions != null ? abortedTransactions.hashCode() : 0); - result = 31 * result + (records != null ? records.hashCode() : 0); - return result; + return this.partitionResponse.hashCode(); } @Override public String toString() { - return "(error=" + error + - ", highWaterMark=" + highWatermark + - ", lastStableOffset = " + lastStableOffset + - ", logStartOffset = " + logStartOffset + - ", preferredReadReplica = " + preferredReadReplica.map(Object::toString).orElse("absent") + - ", abortedTransactions = " + abortedTransactions + - ", recordsSizeInBytes=" + records.sizeInBytes() + ")"; + return "(error=" + error() + + ", highWaterMark=" + highWatermark() + + ", lastStableOffset = " + lastStableOffset() + + ", logStartOffset = " + logStartOffset() + + ", preferredReadReplica = " + preferredReadReplica().map(Object::toString).orElse("absent") + + ", abortedTransactions = " + abortedTransactions() + + ", recordsSizeInBytes=" + records().sizeInBytes() + ")"; + } + + public Errors error() { + return error; + } + + public long highWatermark() { + return partitionResponse.partitionHeader().highWatermark(); + } + + public long lastStableOffset() { + return partitionResponse.partitionHeader().lastStableOffset(); } + public long logStartOffset() { + return partitionResponse.partitionHeader().logStartOffset(); + } + + public Optional preferredReadReplica() { + return preferredReplica; + } + + public List abortedTransactions() { + return abortedTransactions; + } + + @SuppressWarnings("unchecked") + public T records() { + return (T) partitionResponse.recordSet(); + } } /** @@ -366,225 +254,128 @@ public FetchResponse(Errors error, LinkedHashMap> responseData, int throttleTimeMs, int sessionId) { - this.error = error; - this.responseData = responseData; - this.throttleTimeMs = throttleTimeMs; - this.sessionId = sessionId; + this.data = toMessage(throttleTimeMs, error, responseData.entrySet().iterator(), sessionId); + this.responseDataMap = responseData; } - public static FetchResponse parse(Struct struct) { - LinkedHashMap> responseData = new LinkedHashMap<>(); - for (Object topicResponseObj : struct.getArray(RESPONSES_KEY_NAME)) { - Struct topicResponse = (Struct) topicResponseObj; - String topic = topicResponse.get(TOPIC_NAME); - for (Object partitionResponseObj : topicResponse.getArray(PARTITIONS_KEY_NAME)) { - Struct partitionResponse = (Struct) partitionResponseObj; - Struct partitionResponseHeader = partitionResponse.getStruct(PARTITION_HEADER_KEY_NAME); - int partition = partitionResponseHeader.get(PARTITION_ID); - Errors error = Errors.forCode(partitionResponseHeader.get(ERROR_CODE)); - long highWatermark = partitionResponseHeader.get(HIGH_WATERMARK); - long lastStableOffset = partitionResponseHeader.getOrElse(LAST_STABLE_OFFSET, INVALID_LAST_STABLE_OFFSET); - long logStartOffset = partitionResponseHeader.getOrElse(LOG_START_OFFSET, INVALID_LOG_START_OFFSET); - Optional preferredReadReplica = Optional.of( - partitionResponseHeader.getOrElse(PREFERRED_READ_REPLICA, INVALID_PREFERRED_REPLICA_ID) - ).filter(Predicate.isEqual(INVALID_PREFERRED_REPLICA_ID).negate()); - - BaseRecords baseRecords = partitionResponse.getRecords(RECORD_SET_KEY_NAME); - if (!(baseRecords instanceof MemoryRecords)) - throw new IllegalStateException("Unknown records type found: " + baseRecords.getClass()); - MemoryRecords records = (MemoryRecords) baseRecords; - - List abortedTransactions = null; - if (partitionResponseHeader.hasField(ABORTED_TRANSACTIONS_KEY_NAME)) { - Object[] abortedTransactionsArray = partitionResponseHeader.getArray(ABORTED_TRANSACTIONS_KEY_NAME); - if (abortedTransactionsArray != null) { - abortedTransactions = new ArrayList<>(abortedTransactionsArray.length); - for (Object abortedTransactionObj : abortedTransactionsArray) { - Struct abortedTransactionStruct = (Struct) abortedTransactionObj; - long producerId = abortedTransactionStruct.get(PRODUCER_ID); - long firstOffset = abortedTransactionStruct.get(FIRST_OFFSET); - abortedTransactions.add(new AbortedTransaction(producerId, firstOffset)); - } - } - } - - PartitionData partitionData = new PartitionData<>(error, highWatermark, lastStableOffset, - logStartOffset, preferredReadReplica, abortedTransactions, records); - responseData.put(new TopicPartition(topic, partition), partitionData); - } - } - return new FetchResponse<>(Errors.forCode(struct.getOrElse(ERROR_CODE, (short) 0)), responseData, - struct.getOrElse(THROTTLE_TIME_MS, DEFAULT_THROTTLE_TIME), struct.getOrElse(SESSION_ID, INVALID_SESSION_ID)); + public FetchResponse(FetchResponseData fetchResponseData) { + this.data = fetchResponseData; + this.responseDataMap = toResponseDataMap(fetchResponseData); } @Override public Struct toStruct(short version) { - return toStruct(version, throttleTimeMs, error, responseData.entrySet().iterator(), sessionId); + return data.toStruct(version); } @Override - protected Send toSend(String dest, ResponseHeader responseHeader, short apiVersion) { - Struct responseHeaderStruct = responseHeader.toStruct(); - Struct responseBodyStruct = toStruct(apiVersion); - - // write the total size and the response header - ByteBuffer buffer = ByteBuffer.allocate(responseHeaderStruct.sizeOf() + 4); - buffer.putInt(responseHeaderStruct.sizeOf() + responseBodyStruct.sizeOf()); - responseHeaderStruct.writeTo(buffer); + public Send toSend(String dest, ResponseHeader responseHeader, short apiVersion) { + // Generate the Sends for the response fields and records + ArrayDeque sends = new ArrayDeque<>(); + ObjectSerializationCache cache = new ObjectSerializationCache(); + int totalRecordSize = data.responses().stream() + .flatMap(fetchableTopicResponse -> fetchableTopicResponse.partitionResponses().stream()) + .mapToInt(fetchablePartitionResponse -> fetchablePartitionResponse.recordSet().sizeInBytes()) + .sum(); + int totalMessageSize = data.size(cache, apiVersion); + + RecordsWritable writer = new RecordsWritable(dest, totalMessageSize - totalRecordSize, sends::add); + data.write(writer, cache, apiVersion); + writer.flush(); + + // Compute the total size of all the Sends and write it out along with the header in the first Send + ResponseHeaderData responseHeaderData = responseHeader.data(); + + int headerSize = responseHeaderData.size(cache, responseHeader.headerVersion()); + int bodySize = Math.toIntExact(sends.stream().mapToLong(Send::size).sum()); + + ByteBuffer buffer = ByteBuffer.allocate(headerSize + 4); + ByteBufferAccessor headerWriter = new ByteBufferAccessor(buffer); + + // Write out the size and header + buffer.putInt(headerSize + bodySize); + responseHeaderData.write(headerWriter, cache, responseHeader.headerVersion()); + + // Rewind the buffer and set this the first Send in the MultiRecordsSend buffer.rewind(); + sends.addFirst(new ByteBufferSend(dest, buffer)); - Queue sends = new ArrayDeque<>(); - sends.add(new ByteBufferSend(dest, buffer)); - addResponseData(responseBodyStruct, throttleTimeMs, dest, sends); return new MultiRecordsSend(dest, sends); } public Errors error() { - return error; + return Errors.forCode(data.errorCode()); } public LinkedHashMap> responseData() { - return responseData; + return responseDataMap; } @Override public int throttleTimeMs() { - return this.throttleTimeMs; + return data.throttleTimeMs(); } public int sessionId() { - return sessionId; + return data.sessionId(); } @Override public Map errorCounts() { Map errorCounts = new HashMap<>(); - responseData.values().forEach(response -> - updateErrorCounts(errorCounts, response.error) + responseDataMap.values().forEach(response -> + updateErrorCounts(errorCounts, response.error()) ); return errorCounts; } public static FetchResponse parse(ByteBuffer buffer, short version) { - return parse(ApiKeys.FETCH.responseSchema(version).read(buffer)); - } - - private static void addResponseData(Struct struct, int throttleTimeMs, String dest, Queue sends) { - Object[] allTopicData = struct.getArray(RESPONSES_KEY_NAME); - - if (struct.hasField(ERROR_CODE)) { - ByteBuffer buffer = ByteBuffer.allocate(14); - buffer.putInt(throttleTimeMs); - buffer.putShort(struct.get(ERROR_CODE)); - buffer.putInt(struct.get(SESSION_ID)); - buffer.putInt(allTopicData.length); - buffer.rewind(); - sends.add(new ByteBufferSend(dest, buffer)); - } else if (struct.hasField(THROTTLE_TIME_MS)) { - ByteBuffer buffer = ByteBuffer.allocate(8); - buffer.putInt(throttleTimeMs); - buffer.putInt(allTopicData.length); - buffer.rewind(); - sends.add(new ByteBufferSend(dest, buffer)); - } else { - ByteBuffer buffer = ByteBuffer.allocate(4); - buffer.putInt(allTopicData.length); - buffer.rewind(); - sends.add(new ByteBufferSend(dest, buffer)); - } - - for (Object topicData : allTopicData) - addTopicData(dest, sends, (Struct) topicData); + FetchResponseData fetchResponseData = new FetchResponseData(); + RecordsReadable reader = new RecordsReadable(buffer); + fetchResponseData.read(reader, version); + return new FetchResponse<>(fetchResponseData); } - private static void addTopicData(String dest, Queue sends, Struct topicData) { - String topic = topicData.get(TOPIC_NAME); - Object[] allPartitionData = topicData.getArray(PARTITIONS_KEY_NAME); - - // include the topic header and the count for the number of partitions - ByteBuffer buffer = ByteBuffer.allocate(STRING.sizeOf(topic) + 4); - STRING.write(buffer, topic); - buffer.putInt(allPartitionData.length); - buffer.rewind(); - sends.add(new ByteBufferSend(dest, buffer)); - - for (Object partitionData : allPartitionData) - addPartitionData(dest, sends, (Struct) partitionData); + @SuppressWarnings("unchecked") + private static LinkedHashMap> toResponseDataMap( + FetchResponseData message) { + LinkedHashMap> responseMap = new LinkedHashMap<>(); + message.responses().forEach(topicResponse -> { + topicResponse.partitionResponses().forEach(partitionResponse -> { + FetchResponseData.PartitionHeader partitionHeader = partitionResponse.partitionHeader(); + TopicPartition tp = new TopicPartition(topicResponse.topic(), partitionHeader.partition()); + PartitionData partitionData = new PartitionData<>(partitionResponse); + responseMap.put(tp, partitionData); + }); + }); + return responseMap; } - private static void addPartitionData(String dest, Queue sends, Struct partitionData) { - Struct header = partitionData.getStruct(PARTITION_HEADER_KEY_NAME); - BaseRecords records = partitionData.getRecords(RECORD_SET_KEY_NAME); + private static FetchResponseData toMessage(int throttleTimeMs, Errors error, + Iterator>> partIterator, + int sessionId) { + FetchResponseData message = new FetchResponseData(); + message.setThrottleTimeMs(throttleTimeMs); + message.setErrorCode(error.code()); + message.setSessionId(sessionId); - // include the partition header and the size of the record set - ByteBuffer buffer = ByteBuffer.allocate(header.sizeOf() + 4); - header.writeTo(buffer); - buffer.putInt(records.sizeInBytes()); - buffer.rewind(); - sends.add(new ByteBufferSend(dest, buffer)); - - // finally the send for the record set itself - RecordsSend recordsSend = records.toSend(dest); - if (recordsSend.size() > 0) - sends.add(recordsSend); - } - - private static Struct toStruct(short version, int throttleTimeMs, Errors error, - Iterator>> partIterator, - int sessionId) { - Struct struct = new Struct(ApiKeys.FETCH.responseSchema(version)); - struct.setIfExists(THROTTLE_TIME_MS, throttleTimeMs); - struct.setIfExists(ERROR_CODE, error.code()); - struct.setIfExists(SESSION_ID, sessionId); + List topicResponseList = new ArrayList<>(); List>> topicsData = FetchRequest.TopicAndPartitionData.batchByTopic(partIterator); - List topicArray = new ArrayList<>(); - for (FetchRequest.TopicAndPartitionData> topicEntry: topicsData) { - Struct topicData = struct.instance(RESPONSES_KEY_NAME); - topicData.set(TOPIC_NAME, topicEntry.topic); - List partitionArray = new ArrayList<>(); - for (Map.Entry> partitionEntry : topicEntry.partitions.entrySet()) { - PartitionData fetchPartitionData = partitionEntry.getValue(); - short errorCode = fetchPartitionData.error.code(); - // If consumer sends FetchRequest V5 or earlier, the client library is not guaranteed to recognize the error code - // for KafkaStorageException. In this case the client library will translate KafkaStorageException to - // UnknownServerException which is not retriable. We can ensure that consumer will update metadata and retry - // by converting the KafkaStorageException to NotLeaderOrFollowerException in the response if FetchRequest version <= 5 - if (errorCode == Errors.KAFKA_STORAGE_ERROR.code() && version <= 5) - errorCode = Errors.NOT_LEADER_OR_FOLLOWER.code(); - Struct partitionData = topicData.instance(PARTITIONS_KEY_NAME); - Struct partitionDataHeader = partitionData.instance(PARTITION_HEADER_KEY_NAME); - partitionDataHeader.set(PARTITION_ID, partitionEntry.getKey()); - partitionDataHeader.set(ERROR_CODE, errorCode); - partitionDataHeader.set(HIGH_WATERMARK, fetchPartitionData.highWatermark); - - if (partitionDataHeader.hasField(LAST_STABLE_OFFSET)) { - partitionDataHeader.set(LAST_STABLE_OFFSET, fetchPartitionData.lastStableOffset); - - if (fetchPartitionData.abortedTransactions == null) { - partitionDataHeader.set(ABORTED_TRANSACTIONS_KEY_NAME, null); - } else { - List abortedTransactionStructs = new ArrayList<>(fetchPartitionData.abortedTransactions.size()); - for (AbortedTransaction abortedTransaction : fetchPartitionData.abortedTransactions) { - Struct abortedTransactionStruct = partitionDataHeader.instance(ABORTED_TRANSACTIONS_KEY_NAME); - abortedTransactionStruct.set(PRODUCER_ID, abortedTransaction.producerId); - abortedTransactionStruct.set(FIRST_OFFSET, abortedTransaction.firstOffset); - abortedTransactionStructs.add(abortedTransactionStruct); - } - partitionDataHeader.set(ABORTED_TRANSACTIONS_KEY_NAME, abortedTransactionStructs.toArray()); - } - } - partitionDataHeader.setIfExists(LOG_START_OFFSET, fetchPartitionData.logStartOffset); - partitionDataHeader.setIfExists(PREFERRED_READ_REPLICA, fetchPartitionData.preferredReadReplica.orElse(-1)); - partitionData.set(PARTITION_HEADER_KEY_NAME, partitionDataHeader); - partitionData.set(RECORD_SET_KEY_NAME, fetchPartitionData.records); - partitionArray.add(partitionData); - } - topicData.set(PARTITIONS_KEY_NAME, partitionArray.toArray()); - topicArray.add(topicData); - } - struct.set(RESPONSES_KEY_NAME, topicArray.toArray()); - return struct; + topicsData.forEach(partitionDataTopicAndPartitionData -> { + List partitionResponses = new ArrayList<>(); + partitionDataTopicAndPartitionData.partitions.forEach((partitionId, partitionData) -> { + // Since PartitionData alone doesn't know the partition ID, we set it here + partitionData.partitionResponse.partitionHeader().setPartition(partitionId); + partitionResponses.add(partitionData.partitionResponse); + }); + topicResponseList.add(new FetchResponseData.FetchableTopicResponse() + .setTopic(partitionDataTopicAndPartitionData.topic) + .setPartitionResponses(partitionResponses)); + }); + + message.setResponses(topicResponseList); + return message; } /** @@ -598,7 +389,9 @@ public static int sizeOf(short version, Iterator>> partIterator) { // Since the throttleTimeMs and metadata field sizes are constant and fixed, we can // use arbitrary values here without affecting the result. - return 4 + toStruct(version, 0, Errors.NONE, partIterator, INVALID_SESSION_ID).sizeOf(); + FetchResponseData data = toMessage(0, Errors.NONE, partIterator, INVALID_SESSION_ID); + ObjectSerializationCache cache = new ObjectSerializationCache(); + return 4 + data.size(cache, version); } @Override diff --git a/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrRequest.java index 270069aef3c58..2e6b9560baeea 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrRequest.java @@ -175,8 +175,7 @@ public List liveLeaders() { return Collections.unmodifiableList(data.liveLeaders()); } - // Visible for testing - LeaderAndIsrRequestData data() { + public LeaderAndIsrRequestData data() { return data; } diff --git a/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaRequest.java index 0f79f2d4206af..8ab8cc38ecad7 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaRequest.java @@ -214,8 +214,7 @@ public static StopReplicaRequest parse(ByteBuffer buffer, short version) { return new StopReplicaRequest(ApiKeys.STOP_REPLICA.parseRequest(version, buffer), version); } - // Visible for testing - StopReplicaRequestData data() { + public StopReplicaRequestData data() { return data; } diff --git a/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java index 49f962d28938e..f17ea218d3aa0 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java @@ -209,8 +209,7 @@ protected Struct toStruct() { return data.toStruct(version()); } - // Visible for testing - UpdateMetadataRequestData data() { + public UpdateMetadataRequestData data() { return data; } diff --git a/clients/src/main/resources/common/message/FetchRequest.json b/clients/src/main/resources/common/message/FetchRequest.json index 727618c3382a5..364bd73a67b9a 100644 --- a/clients/src/main/resources/common/message/FetchRequest.json +++ b/clients/src/main/resources/common/message/FetchRequest.json @@ -55,35 +55,35 @@ "about": "The minimum bytes to accumulate in the response." }, { "name": "MaxBytes", "type": "int32", "versions": "3+", "default": "0x7fffffff", "ignorable": true, "about": "The maximum bytes to fetch. See KIP-74 for cases where this limit may not be honored." }, - { "name": "IsolationLevel", "type": "int8", "versions": "4+", "default": "0", "ignorable": false, + { "name": "IsolationLevel", "type": "int8", "versions": "4+", "default": "0", "ignorable": true, "about": "This setting controls the visibility of transactional records. Using READ_UNCOMMITTED (isolation_level = 0) makes all records visible. With READ_COMMITTED (isolation_level = 1), non-transactional and COMMITTED transactional records are visible. To be more concrete, READ_COMMITTED returns all data from offsets smaller than the current LSO (last stable offset), and enables the inclusion of the list of aborted transactions in the result, which allows consumers to discard ABORTED transactional records" }, - { "name": "SessionId", "type": "int32", "versions": "7+", "default": "0", "ignorable": false, + { "name": "SessionId", "type": "int32", "versions": "7+", "default": "0", "ignorable": true, "about": "The fetch session ID." }, - { "name": "SessionEpoch", "type": "int32", "versions": "7+", "default": "-1", "ignorable": false, - "about": "The fetch session epoch, which is used for ordering requests in a session" }, + { "name": "SessionEpoch", "type": "int32", "versions": "7+", "default": "-1", "ignorable": true, + "about": "The fetch session epoch, which is used for ordering requests in a session." }, { "name": "Topics", "type": "[]FetchTopic", "versions": "0+", "about": "The topics to fetch.", "fields": [ - { "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName", + { "name": "Topic", "type": "string", "versions": "0+", "entityType": "topicName", "about": "The name of the topic to fetch." }, - { "name": "FetchPartitions", "type": "[]FetchPartition", "versions": "0+", + { "name": "Partitions", "type": "[]FetchPartition", "versions": "0+", "about": "The partitions to fetch.", "fields": [ - { "name": "PartitionIndex", "type": "int32", "versions": "0+", + { "name": "Partition", "type": "int32", "versions": "0+", "about": "The partition index." }, { "name": "CurrentLeaderEpoch", "type": "int32", "versions": "9+", "default": "-1", "ignorable": true, "about": "The current leader epoch of the partition." }, { "name": "FetchOffset", "type": "int64", "versions": "0+", "about": "The message offset." }, - { "name": "LogStartOffset", "type": "int64", "versions": "5+", "default": "-1", "ignorable": false, + { "name": "LogStartOffset", "type": "int64", "versions": "5+", "default": "-1", "ignorable": true, "about": "The earliest available offset of the follower replica. The field is only used when the request is sent by the follower."}, - { "name": "MaxBytes", "type": "int32", "versions": "0+", + { "name": "PartitionMaxBytes", "type": "int32", "versions": "0+", "about": "The maximum bytes to fetch from this partition. See KIP-74 for cases where this limit may not be honored." } ]} ]}, - { "name": "Forgotten", "type": "[]ForgottenTopic", "versions": "7+", "ignorable": false, + { "name": "ForgottenTopicsData", "type": "[]ForgottenTopic", "versions": "7+", "ignorable": false, "about": "In an incremental fetch request, the partitions to remove.", "fields": [ - { "name": "Name", "type": "string", "versions": "7+", "entityType": "topicName", + { "name": "Topic", "type": "string", "versions": "7+", "entityType": "topicName", "about": "The partition name." }, - { "name": "ForgottenPartitionIndexes", "type": "[]int32", "versions": "7+", + { "name": "Partitions", "type": "[]int32", "versions": "7+", "about": "The partitions indexes to forget." } ]}, { "name": "RackId", "type": "string", "versions": "11+", "default": "", "ignorable": true, diff --git a/clients/src/main/resources/common/message/FetchResponse.json b/clients/src/main/resources/common/message/FetchResponse.json index 0fe98ade2f5f7..9f64c2428f763 100644 --- a/clients/src/main/resources/common/message/FetchResponse.json +++ b/clients/src/main/resources/common/message/FetchResponse.json @@ -43,37 +43,39 @@ "fields": [ { "name": "ThrottleTimeMs", "type": "int32", "versions": "1+", "ignorable": true, "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." }, - { "name": "ErrorCode", "type": "int16", "versions": "7+", "ignorable": false, + { "name": "ErrorCode", "type": "int16", "versions": "7+", "ignorable": true, "about": "The top level response error code." }, { "name": "SessionId", "type": "int32", "versions": "7+", "default": "0", "ignorable": false, "about": "The fetch session ID, or 0 if this is not part of a fetch session." }, - { "name": "Topics", "type": "[]FetchableTopicResponse", "versions": "0+", + { "name": "Responses", "type": "[]FetchableTopicResponse", "versions": "0+", "about": "The response topics.", "fields": [ - { "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName", + { "name": "Topic", "type": "string", "versions": "0+", "entityType": "topicName", "about": "The topic name." }, - { "name": "Partitions", "type": "[]FetchablePartitionResponse", "versions": "0+", + { "name": "PartitionResponses", "type": "[]FetchablePartitionResponse", "versions": "0+", "about": "The topic partitions.", "fields": [ - { "name": "PartitionIndex", "type": "int32", "versions": "0+", - "about": "The partiiton index." }, - { "name": "ErrorCode", "type": "int16", "versions": "0+", - "about": "The error code, or 0 if there was no fetch error." }, - { "name": "HighWatermark", "type": "int64", "versions": "0+", - "about": "The current high water mark." }, - { "name": "LastStableOffset", "type": "int64", "versions": "4+", "default": "-1", "ignorable": true, - "about": "The last stable offset (or LSO) of the partition. This is the last offset such that the state of all transactional records prior to this offset have been decided (ABORTED or COMMITTED)" }, - { "name": "LogStartOffset", "type": "int64", "versions": "5+", "default": "-1", "ignorable": true, - "about": "The current log start offset." }, - { "name": "Aborted", "type": "[]AbortedTransaction", "versions": "4+", "nullableVersions": "4+", "ignorable": false, - "about": "The aborted transactions.", "fields": [ - { "name": "ProducerId", "type": "int64", "versions": "4+", "entityType": "producerId", - "about": "The producer id associated with the aborted transaction." }, - { "name": "FirstOffset", "type": "int64", "versions": "4+", - "about": "The first offset in the aborted transaction." } + { "name": "PartitionHeader", "type": "PartitionHeader", "versions": "0+", + "fields": [ + { "name": "Partition", "type": "int32", "versions": "0+", + "about": "The partition index." }, + { "name": "ErrorCode", "type": "int16", "versions": "0+", + "about": "The error code, or 0 if there was no fetch error." }, + { "name": "HighWatermark", "type": "int64", "versions": "0+", + "about": "The current high water mark." }, + { "name": "LastStableOffset", "type": "int64", "versions": "4+", "default": "-1", "ignorable": true, + "about": "The last stable offset (or LSO) of the partition. This is the last offset such that the state of all transactional records prior to this offset have been decided (ABORTED or COMMITTED)" }, + { "name": "LogStartOffset", "type": "int64", "versions": "5+", "default": "-1", "ignorable": true, + "about": "The current log start offset." }, + { "name": "AbortedTransactions", "type": "[]AbortedTransaction", "versions": "4+", "nullableVersions": "4+", "ignorable": true, + "about": "The aborted transactions.", "fields": [ + { "name": "ProducerId", "type": "int64", "versions": "4+", "entityType": "producerId", + "about": "The producer id associated with the aborted transaction." }, + { "name": "FirstOffset", "type": "int64", "versions": "4+", + "about": "The first offset in the aborted transaction." } + ]}, + { "name": "PreferredReadReplica", "type": "int32", "versions": "11+", "default": "-1", "ignorable": false, + "about": "The preferred read replica for the consumer to use on its next fetch request"} ]}, - { "name": "PreferredReadReplica", "type": "int32", "versions": "11+", "ignorable": true, - "about": "The preferred read replica for the consumer to use on its next fetch request"}, - { "name": "Records", "type": "bytes", "versions": "0+", "nullableVersions": "0+", - "about": "The record data." } + { "name": "RecordSet", "type": "records", "versions": "0+", "nullableVersions": "0+", "about": "The record data."} ]} ]} ] diff --git a/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java b/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java index bbff0ed5201ac..a3946bd3ca058 100644 --- a/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java +++ b/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java @@ -951,8 +951,8 @@ public void testDefaultValues() { verifyWriteSucceeds((short) 0, new OffsetCommitRequestData().setRetentionTimeMs(123)); verifyWriteRaisesUve((short) 5, "forgotten", - new FetchRequestData().setForgotten(singletonList( - new FetchRequestData.ForgottenTopic().setName("foo")))); + new FetchRequestData().setForgottenTopicsData(singletonList( + new FetchRequestData.ForgottenTopic().setTopic("foo")))); } @Test diff --git a/clients/src/test/java/org/apache/kafka/common/protocol/RecordsWritableTest.java b/clients/src/test/java/org/apache/kafka/common/protocol/RecordsWritableTest.java new file mode 100644 index 0000000000000..937da4d449611 --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/common/protocol/RecordsWritableTest.java @@ -0,0 +1,63 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.kafka.common.protocol; + +import org.apache.kafka.common.network.ByteBufferSend; +import org.apache.kafka.common.network.Send; +import org.junit.Assert; +import org.junit.Test; + +import java.util.ArrayDeque; +import java.util.Queue; + +public class RecordsWritableTest { + @Test + public void testBufferSlice() { + Queue sends = new ArrayDeque<>(); + RecordsWritable writer = new RecordsWritable("dest", 10000 /* enough for tests */, sends::add); + for (int i = 0; i < 4; i++) { + writer.writeInt(i); + } + writer.flush(); + Assert.assertEquals(sends.size(), 1); + ByteBufferSend send = (ByteBufferSend) sends.remove(); + Assert.assertEquals(send.size(), 16); + Assert.assertEquals(send.remaining(), 16); + + // No new data, flush shouldn't do anything + writer.flush(); + Assert.assertEquals(sends.size(), 0); + + // Cause the buffer to expand a few times + for (int i = 0; i < 100; i++) { + writer.writeInt(i); + } + writer.flush(); + Assert.assertEquals(sends.size(), 1); + send = (ByteBufferSend) sends.remove(); + Assert.assertEquals(send.size(), 400); + Assert.assertEquals(send.remaining(), 400); + + writer.writeByte((byte) 5); + writer.flush(); + Assert.assertEquals(sends.size(), 1); + send = (ByteBufferSend) sends.remove(); + Assert.assertEquals(send.size(), 1); + Assert.assertEquals(send.remaining(), 1); + } +} diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java index 723b67118c96a..ae41bf9fd3b9e 100644 --- a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java +++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java @@ -93,6 +93,7 @@ import org.apache.kafka.common.message.EndTxnResponseData; import org.apache.kafka.common.message.ExpireDelegationTokenRequestData; import org.apache.kafka.common.message.ExpireDelegationTokenResponseData; +import org.apache.kafka.common.message.FetchRequestData; import org.apache.kafka.common.message.FindCoordinatorRequestData; import org.apache.kafka.common.message.HeartbeatRequestData; import org.apache.kafka.common.message.HeartbeatResponseData; @@ -144,7 +145,9 @@ import org.apache.kafka.common.network.ListenerName; import org.apache.kafka.common.network.Send; import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.ByteBufferAccessor; import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.protocol.ObjectSerializationCache; import org.apache.kafka.common.protocol.types.SchemaException; import org.apache.kafka.common.protocol.types.Struct; import org.apache.kafka.common.quota.ClientQuotaAlteration; @@ -216,7 +219,7 @@ public void testSerialization() throws Exception { checkErrorResponse(createControlledShutdownRequest(), unknownServerException, true); checkErrorResponse(createControlledShutdownRequest(0), unknownServerException, true); checkRequest(createFetchRequest(4), true); - checkResponse(createFetchResponse(), 4, true); + checkResponse(createFetchResponse(true), 4, true); List toForgetTopics = new ArrayList<>(); toForgetTopics.add(new TopicPartition("foo", 0)); toForgetTopics.add(new TopicPartition("foo", 2)); @@ -224,7 +227,7 @@ public void testSerialization() throws Exception { checkRequest(createFetchRequest(7, new FetchMetadata(123, 456), toForgetTopics), true); checkResponse(createFetchResponse(123), 7, true); checkResponse(createFetchResponse(Errors.FETCH_SESSION_ID_NOT_FOUND, 123), 7, true); - checkErrorResponse(createFetchRequest(4), unknownServerException, true); + checkErrorResponse(createFetchRequest(7), unknownServerException, true); checkRequest(createHeartBeatRequest(), true); checkErrorResponse(createHeartBeatRequest(), unknownServerException, true); checkResponse(createHeartBeatResponse(), 0, true); @@ -509,9 +512,11 @@ public void testResponseHeader() { private void checkOlderFetchVersions() throws Exception { int latestVersion = FETCH.latestVersion(); for (int i = 0; i < latestVersion; ++i) { - checkErrorResponse(createFetchRequest(i), unknownServerException, true); + if (i > 7) { + checkErrorResponse(createFetchRequest(i), unknownServerException, true); + } checkRequest(createFetchRequest(i), true); - checkResponse(createFetchResponse(), i, true); + checkResponse(createFetchResponse(i >= 4), i, true); } } @@ -676,7 +681,7 @@ public void fetchResponseVersionTest() { MemoryRecords records = MemoryRecords.readableRecords(ByteBuffer.allocate(10)); responseData.put(new TopicPartition("test", 0), new FetchResponse.PartitionData<>( Errors.NONE, 1000000, FetchResponse.INVALID_LAST_STABLE_OFFSET, - 0L, Optional.empty(), null, records)); + 0L, Optional.empty(), Collections.emptyList(), records)); FetchResponse v0Response = new FetchResponse<>(Errors.NONE, responseData, 0, INVALID_SESSION_ID); FetchResponse v1Response = new FetchResponse<>(Errors.NONE, responseData, 10, INVALID_SESSION_ID); @@ -717,7 +722,7 @@ public void verifyFetchResponseFullWrites() throws Exception { verifyFetchResponseFullWrite(FETCH.latestVersion(), createFetchResponse(Errors.FETCH_SESSION_ID_NOT_FOUND, 123)); for (short version = 0; version <= FETCH.latestVersion(); version++) { - verifyFetchResponseFullWrite(version, createFetchResponse()); + verifyFetchResponseFullWrite(version, createFetchResponse(version >= 4)); } } @@ -789,7 +794,7 @@ public void testCreateTopicRequestV3FailsIfNoPartitionsOrReplicas() { public void testFetchRequestMaxBytesOldVersions() throws Exception { final short version = 1; FetchRequest fr = createFetchRequest(version); - FetchRequest fr2 = new FetchRequest(fr.toStruct(), version); + FetchRequest fr2 = new FetchRequest(new FetchRequestData(fr.toStruct(), version), version); assertEquals(fr2.maxBytes(), fr.maxBytes()); } @@ -819,6 +824,24 @@ public void testFetchRequestWithMetadata() throws Exception { assertEquals(request.isolationLevel(), deserialized.isolationLevel()); } + @Test + public void testFetchRequestCompat() { + Map fetchData = new HashMap<>(); + fetchData.put(new TopicPartition("test", 0), new FetchRequest.PartitionData(100, 2, 100, Optional.of(42))); + FetchRequest req = FetchRequest.Builder + .forConsumer(100, 100, fetchData) + .metadata(new FetchMetadata(10, 20)) + .isolationLevel(IsolationLevel.READ_COMMITTED) + .build((short) 2); + + FetchRequestData data = req.data(); + ObjectSerializationCache cache = new ObjectSerializationCache(); + int size = data.size(cache, (short) 2); + + ByteBufferAccessor writer = new ByteBufferAccessor(ByteBuffer.allocate(size)); + data.write(writer, cache, (short) 2); + } + @Test public void testJoinGroupRequestVersion0RebalanceTimeout() { final short version = 0; @@ -968,30 +991,30 @@ private FindCoordinatorResponse createFindCoordinatorResponse() { private FetchRequest createFetchRequest(int version, FetchMetadata metadata, List toForget) { LinkedHashMap fetchData = new LinkedHashMap<>(); - fetchData.put(new TopicPartition("test1", 0), new FetchRequest.PartitionData(100, 0L, - 1000000, Optional.of(15))); - fetchData.put(new TopicPartition("test2", 0), new FetchRequest.PartitionData(200, 0L, - 1000000, Optional.of(25))); + fetchData.put(new TopicPartition("test1", 0), new FetchRequest.PartitionData(100, -1L, + 1000000, Optional.empty())); + fetchData.put(new TopicPartition("test2", 0), new FetchRequest.PartitionData(200, -1L, + 1000000, Optional.empty())); return FetchRequest.Builder.forConsumer(100, 100000, fetchData). metadata(metadata).setMaxBytes(1000).toForget(toForget).build((short) version); } private FetchRequest createFetchRequest(int version, IsolationLevel isolationLevel) { LinkedHashMap fetchData = new LinkedHashMap<>(); - fetchData.put(new TopicPartition("test1", 0), new FetchRequest.PartitionData(100, 0L, - 1000000, Optional.of(15))); - fetchData.put(new TopicPartition("test2", 0), new FetchRequest.PartitionData(200, 0L, - 1000000, Optional.of(25))); + fetchData.put(new TopicPartition("test1", 0), new FetchRequest.PartitionData(100, -1L, + 1000000, Optional.empty())); + fetchData.put(new TopicPartition("test2", 0), new FetchRequest.PartitionData(200, -1L, + 1000000, Optional.empty())); return FetchRequest.Builder.forConsumer(100, 100000, fetchData). isolationLevel(isolationLevel).setMaxBytes(1000).build((short) version); } private FetchRequest createFetchRequest(int version) { LinkedHashMap fetchData = new LinkedHashMap<>(); - fetchData.put(new TopicPartition("test1", 0), new FetchRequest.PartitionData(100, 0L, - 1000000, Optional.of(15))); - fetchData.put(new TopicPartition("test2", 0), new FetchRequest.PartitionData(200, 0L, - 1000000, Optional.of(25))); + fetchData.put(new TopicPartition("test1", 0), new FetchRequest.PartitionData(100, -1L, + 1000000, Optional.empty())); + fetchData.put(new TopicPartition("test2", 0), new FetchRequest.PartitionData(200, -1L, + 1000000, Optional.empty())); return FetchRequest.Builder.forConsumer(100, 100000, fetchData).setMaxBytes(1000).build((short) version); } @@ -1003,7 +1026,7 @@ private FetchResponse createFetchResponse(int sessionId) { LinkedHashMap> responseData = new LinkedHashMap<>(); MemoryRecords records = MemoryRecords.withRecords(CompressionType.NONE, new SimpleRecord("blah".getBytes())); responseData.put(new TopicPartition("test", 0), new FetchResponse.PartitionData<>(Errors.NONE, - 1000000, FetchResponse.INVALID_LAST_STABLE_OFFSET, 0L, Optional.empty(), null, records)); + 1000000, FetchResponse.INVALID_LAST_STABLE_OFFSET, 0L, Optional.empty(), Collections.emptyList(), records)); List abortedTransactions = Collections.singletonList( new FetchResponse.AbortedTransaction(234L, 999L)); responseData.put(new TopicPartition("test", 1), new FetchResponse.PartitionData<>(Errors.NONE, @@ -1011,14 +1034,18 @@ private FetchResponse createFetchResponse(int sessionId) { return new FetchResponse<>(Errors.NONE, responseData, 25, sessionId); } - private FetchResponse createFetchResponse() { + private FetchResponse createFetchResponse(boolean includeAborted) { LinkedHashMap> responseData = new LinkedHashMap<>(); MemoryRecords records = MemoryRecords.withRecords(CompressionType.NONE, new SimpleRecord("blah".getBytes())); + responseData.put(new TopicPartition("test", 0), new FetchResponse.PartitionData<>(Errors.NONE, - 1000000, FetchResponse.INVALID_LAST_STABLE_OFFSET, 0L, Optional.empty(), null, records)); + 1000000, FetchResponse.INVALID_LAST_STABLE_OFFSET, 0L, Optional.empty(), Collections.emptyList(), records)); - List abortedTransactions = Collections.singletonList( - new FetchResponse.AbortedTransaction(234L, 999L)); + List abortedTransactions = Collections.emptyList(); + if (includeAborted) { + abortedTransactions = Collections.singletonList( + new FetchResponse.AbortedTransaction(234L, 999L)); + } responseData.put(new TopicPartition("test", 1), new FetchResponse.PartitionData<>(Errors.NONE, 1000000, FetchResponse.INVALID_LAST_STABLE_OFFSET, 0L, Optional.empty(), abortedTransactions, MemoryRecords.EMPTY)); diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index a7397664ca4a7..186bc73997bf6 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/core/src/main/scala/kafka/server/KafkaApis.scala @@ -673,6 +673,18 @@ class KafkaApis(val requestChannel: RequestChannel, } } + def maybeDownConvertStorageError(error: Errors, version: Short): Errors = { + // If consumer sends FetchRequest V5 or earlier, the client library is not guaranteed to recognize the error code + // for KafkaStorageException. In this case the client library will translate KafkaStorageException to + // UnknownServerException which is not retriable. We can ensure that consumer will update metadata and retry + // by converting the KafkaStorageException to NotLeaderForPartitionException in the response if FetchRequest version <= 5 + if (error == Errors.KAFKA_STORAGE_ERROR && versionId <= 5) { + Errors.NOT_LEADER_OR_FOLLOWER + } else { + error + } + } + def maybeConvertFetchedData(tp: TopicPartition, partitionData: FetchResponse.PartitionData[Records]): FetchResponse.PartitionData[BaseRecords] = { val logConfig = replicaManager.getLogConfig(tp) @@ -712,7 +724,8 @@ class KafkaApis(val requestChannel: RequestChannel, // as possible. With KIP-283, we have the ability to lazily down-convert in a chunked manner. The lazy, chunked // down-conversion always guarantees that at least one batch of messages is down-converted and sent out to the // client. - new FetchResponse.PartitionData[BaseRecords](partitionData.error, partitionData.highWatermark, + val error = maybeDownConvertStorageError(partitionData.error, versionId) + new FetchResponse.PartitionData[BaseRecords](error, partitionData.highWatermark, partitionData.lastStableOffset, partitionData.logStartOffset, partitionData.preferredReadReplica, partitionData.abortedTransactions, new LazyDownConversionRecords(tp, unconvertedRecords, magic, fetchContext.getFetchOffset(tp).get, time)) @@ -722,10 +735,13 @@ class KafkaApis(val requestChannel: RequestChannel, errorResponse(Errors.UNSUPPORTED_COMPRESSION_TYPE) } } - case None => new FetchResponse.PartitionData[BaseRecords](partitionData.error, partitionData.highWatermark, - partitionData.lastStableOffset, partitionData.logStartOffset, - partitionData.preferredReadReplica, partitionData.abortedTransactions, - unconvertedRecords) + case None => { + val error = maybeDownConvertStorageError(partitionData.error, versionId) + new FetchResponse.PartitionData[BaseRecords](error, partitionData.highWatermark, + partitionData.lastStableOffset, partitionData.logStartOffset, + partitionData.preferredReadReplica, partitionData.abortedTransactions, + unconvertedRecords) + } } } } @@ -739,7 +755,8 @@ class KafkaApis(val requestChannel: RequestChannel, val lastStableOffset = data.lastStableOffset.getOrElse(FetchResponse.INVALID_LAST_STABLE_OFFSET) if (data.isReassignmentFetch) reassigningPartitions.add(tp) - partitions.put(tp, new FetchResponse.PartitionData(data.error, data.highWatermark, lastStableOffset, + val error = maybeDownConvertStorageError(data.error, versionId) + partitions.put(tp, new FetchResponse.PartitionData(error, data.highWatermark, lastStableOffset, data.logStartOffset, data.preferredReadReplica.map(int2Integer).asJava, abortedTransactions, data.records)) } diff --git a/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala b/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala index 6561a487ebc49..33a63f0af9180 100644 --- a/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala @@ -569,7 +569,8 @@ class FetchRequestTest extends BaseRequestTest { // zstd compressed record raises UNSUPPORTED_COMPRESSION_TYPE error. val req0 = new FetchRequest.Builder(0, 1, -1, Int.MaxValue, 0, createPartitionMap(300, Seq(topicPartition), Map.empty)) - .setMaxBytes(800).build() + .setMaxBytes(800) + .build() val res0 = sendFetchRequest(leaderId, req0) val data0 = res0.responseData.get(topicPartition) diff --git a/generator/src/main/java/org/apache/kafka/message/FieldSpec.java b/generator/src/main/java/org/apache/kafka/message/FieldSpec.java index d2afc171f6da3..c0b280bc6afb5 100644 --- a/generator/src/main/java/org/apache/kafka/message/FieldSpec.java +++ b/generator/src/main/java/org/apache/kafka/message/FieldSpec.java @@ -56,7 +56,7 @@ public final class FieldSpec { private final Optional tag; - private boolean zeroCopy; + private final boolean zeroCopy; @JsonCreator public FieldSpec(@JsonProperty("name") String name, diff --git a/generator/src/main/java/org/apache/kafka/message/FieldType.java b/generator/src/main/java/org/apache/kafka/message/FieldType.java index 5e0a4c7a74554..aa4523ee46310 100644 --- a/generator/src/main/java/org/apache/kafka/message/FieldType.java +++ b/generator/src/main/java/org/apache/kafka/message/FieldType.java @@ -182,6 +182,31 @@ public String toString() { } } + final class RecordsFieldType implements FieldType { + static final RecordsFieldType INSTANCE = new RecordsFieldType(); + private static final String NAME = "records"; + + @Override + public boolean serializationIsDifferentInFlexibleVersions() { + return true; + } + + @Override + public boolean isRecords() { + return true; + } + + @Override + public boolean canBeNullable() { + return true; + } + + @Override + public String toString() { + return NAME; + } + } + final class StructType implements FieldType { private final String type; @@ -267,6 +292,8 @@ static FieldType parse(String string) { return StringFieldType.INSTANCE; case BytesFieldType.NAME: return BytesFieldType.INSTANCE; + case RecordsFieldType.NAME: + return RecordsFieldType.INSTANCE; default: if (string.startsWith(ARRAY_PREFIX)) { String elementTypeString = string.substring(ARRAY_PREFIX.length()); @@ -323,6 +350,13 @@ default boolean isBytes() { return false; } + /** + * Returns true if this is a records type + */ + default boolean isRecords() { + return false; + } + /** * Returns true if this is a floating point type. */ diff --git a/generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java b/generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java index c51dcc9f01052..137e86b529526 100644 --- a/generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java +++ b/generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java @@ -390,6 +390,9 @@ private String fieldAbstractJavaType(FieldSpec field) { } else { return "byte[]"; } + } else if (field.type() instanceof FieldType.RecordsFieldType) { + headerGenerator.addImport(MessageGenerator.BASE_RECORDS_CLASS); + return "BaseRecords"; } else if (field.type().isStruct()) { return MessageGenerator.capitalizeFirst(field.typeString()); } else if (field.type().isArray()) { @@ -643,7 +646,7 @@ private void generateVariableLengthReader(Versions fieldFlexibleVersions, ifNotMember(__ -> { if (type.isString()) { buffer.printf("%s = _readable.readShort();%n", lengthVar); - } else if (type.isBytes() || type.isArray()) { + } else if (type.isBytes() || type.isArray() || type.isRecords()) { buffer.printf("%s = _readable.readInt();%n", lengthVar); } else { throw new RuntimeException("Can't handle variable length type " + type); @@ -682,6 +685,17 @@ private void generateVariableLengthReader(Versions fieldFlexibleVersions, buffer.printf("_readable.readArray(newBytes);%n"); buffer.printf("%snewBytes%s", assignmentPrefix, assignmentSuffix); } + } else if (type.isRecords()) { + headerGenerator.addImport(MessageGenerator.RECORDS_READABLE_CLASS); + buffer.printf("if (_readable instanceof RecordsReadable) {%n"); + buffer.incrementIndent(); + buffer.printf("%s((RecordsReadable) _readable).readRecords(%s)%s", assignmentPrefix, lengthVar, assignmentSuffix); + buffer.decrementIndent(); + buffer.printf("} else {%n"); + buffer.incrementIndent(); + buffer.printf("throw new RuntimeException(\"Cannot read records from reader of class: \" + _readable.getClass().getSimpleName());%n"); + buffer.decrementIndent(); + buffer.printf("}%n"); } else if (type.isArray()) { FieldType.ArrayType arrayType = (FieldType.ArrayType) type; if (isStructArrayWithKeys) { @@ -964,6 +978,12 @@ private void generateVariableLengthTargetFromJson(Target target, Versions curVer String.format("MessageUtil.jsonNodeToBinary(%s, \"%s\")", target.sourceVariable(), target.humanReadableName()))); } + } else if (target.field().type().isRecords()) { + headerGenerator.addImport(MessageGenerator.BYTE_BUFFER_CLASS); + headerGenerator.addImport(MessageGenerator.MEMORY_RECORDS_CLASS); + buffer.printf("%s;%n", target.assignmentStatement( + String.format("MemoryRecords.readableRecords(ByteBuffer.wrap(MessageUtil.jsonNodeToBinary(%s, \"%s\")))", + target.sourceVariable(), target.humanReadableName()))); } else if (target.field().type().isArray()) { buffer.printf("if (!%s.isArray()) {%n", target.sourceVariable()); buffer.incrementIndent(); @@ -1109,6 +1129,9 @@ private void generateVariableLengthTargetToJson(Target target, Versions versions String.format("new BinaryNode(Arrays.copyOf(%s, %s.length))", target.sourceVariable(), target.sourceVariable()))); } + } else if (target.field().type().isRecords()) { + headerGenerator.addImport(MessageGenerator.BINARY_NODE_CLASS); + buffer.printf("%s;%n", target.assignmentStatement("new BinaryNode(new byte[]{})")); } else if (target.field().type().isArray()) { headerGenerator.addImport(MessageGenerator.ARRAY_NODE_CLASS); headerGenerator.addImport(MessageGenerator.JSON_NODE_FACTORY_CLASS); @@ -1209,6 +1232,8 @@ private String readFieldFromStruct(FieldType type, String name, boolean zeroCopy } else { return String.format("struct.getByteArray(\"%s\")", name); } + } else if (type.isRecords()) { + return String.format("struct.getRecords(\"%s\")", name); } else if (type.isStruct()) { return String.format("new %s((Struct) struct.get(\"%s\"), _version)", type.toString(), name); @@ -1467,6 +1492,8 @@ private void generateVariableLengthWriter(Versions fieldFlexibleVersions, } else { lengthExpression = String.format("%s.length", name); } + } else if (type.isRecords()) { + lengthExpression = String.format("%s.sizeInBytes()", name); } else if (type.isArray()) { lengthExpression = String.format("%s.size()", name); } else { @@ -1499,6 +1526,17 @@ private void generateVariableLengthWriter(Versions fieldFlexibleVersions, } else { buffer.printf("_writable.writeByteArray(%s);%n", name); } + } else if (type.isRecords()) { + headerGenerator.addImport(MessageGenerator.RECORDS_WRITABLE_CLASS); + buffer.printf("if (_writable instanceof RecordsWritable) {%n"); + buffer.incrementIndent(); + buffer.printf("((RecordsWritable) _writable).writeRecords(%s);%n", name); + buffer.decrementIndent(); + buffer.printf("} else {%n"); + buffer.incrementIndent(); + buffer.printf("throw new RuntimeException(\"Cannot write records to writer of class: \" + _writable.getClass().getSimpleName());%n"); + buffer.decrementIndent(); + buffer.printf("}%n"); } else if (type.isArray()) { FieldType.ArrayType arrayType = (FieldType.ArrayType) type; FieldType elementType = arrayType.elementType(); @@ -1658,6 +1696,9 @@ private void generateFieldToStruct(FieldSpec field, Versions versions) { buffer.printf("struct.setByteArray(\"%s\", this.%s);%n", field.snakeCaseName(), field.camelCaseName()); } + } else if (field.type().isRecords()) { + buffer.printf("struct.set(\"%s\", this.%s);%n", + field.snakeCaseName(), field.camelCaseName()); } else if (field.type().isArray()) { IsNullConditional.forField(field). possibleVersions(versions). @@ -1997,6 +2038,8 @@ private void generateVariableLengthFieldSize(FieldSpec field, } else { buffer.printf("_size += _bytesSize;%n"); } + } else if (field.type().isRecords()) { + buffer.printf("_size += %s.sizeInBytes() + 4;%n", field.camelCaseName()); } else if (field.type().isStruct()) { buffer.printf("int size = this.%s.size(_cache, _version);%n", field.camelCaseName()); if (tagged) { @@ -2078,6 +2121,10 @@ private void generateFieldEquals(FieldSpec field) { buffer.printf("if (!Arrays.equals(this.%s, other.%s)) return false;%n", field.camelCaseName(), field.camelCaseName()); } + } else if (field.type().isRecords()) { + headerGenerator.addImport(MessageGenerator.OBJECTS_CLASS); + buffer.printf("if (!Objects.equals(this.%s, other.%s)) return false;%n", + field.camelCaseName(), field.camelCaseName()); } else { buffer.printf("if (%s != other.%s) return false;%n", field.camelCaseName(), field.camelCaseName()); @@ -2121,12 +2168,16 @@ private void generateFieldHashCode(FieldSpec field) { if (field.zeroCopy()) { headerGenerator.addImport(MessageGenerator.OBJECTS_CLASS); buffer.printf("hashCode = 31 * hashCode + Objects.hashCode(%s);%n", - field.camelCaseName()); + field.camelCaseName()); } else { headerGenerator.addImport(MessageGenerator.ARRAYS_CLASS); buffer.printf("hashCode = 31 * hashCode + Arrays.hashCode(%s);%n", field.camelCaseName()); } + } else if (field.type().isRecords()) { + headerGenerator.addImport(MessageGenerator.OBJECTS_CLASS); + buffer.printf("hashCode = 31 * hashCode + Objects.hashCode(%s);%n", + field.camelCaseName()); } else if (field.type().isStruct() || field.type().isArray() || field.type().isString()) { @@ -2180,6 +2231,13 @@ private void generateFieldDuplicate(Target target) { target.sourceVariable()))); }); } + } else if (field.type().isRecords()) { + cond.ifShouldNotBeNull(() -> { + headerGenerator.addImport(MessageGenerator.MEMORY_RECORDS_CLASS); + buffer.printf("%s;%n", target.assignmentStatement( + String.format("MemoryRecords.readableRecords(((MemoryRecords) %s).buffer().duplicate())", + target.sourceVariable()))); + }); } else if (field.type().isStruct()) { cond.ifShouldNotBeNull(() -> buffer.printf("%s;%n", target.assignmentStatement( @@ -2254,6 +2312,9 @@ private void generateFieldToString(String prefix, FieldSpec field) { buffer.printf("+ \"%s%s=\" + Arrays.toString(%s)%n", prefix, field.camelCaseName(), field.camelCaseName()); } + } else if (field.type().isRecords()) { + buffer.printf("+ \"%s%s=\" + %s%n", + prefix, field.camelCaseName(), field.camelCaseName()); } else if (field.type().isStruct() || field.type() instanceof FieldType.UUIDFieldType) { } else if (field.type().isStruct()) { @@ -2397,6 +2458,8 @@ private String fieldDefault(FieldSpec field) { headerGenerator.addImport(MessageGenerator.BYTES_CLASS); return "Bytes.EMPTY"; } + } else if (field.type().isRecords()) { + return "null"; } else if (field.type().isStruct()) { if (!field.defaultString().isEmpty()) { throw new RuntimeException("Invalid default for struct field " + diff --git a/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java b/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java index c84b60bd317f8..74d85b0552442 100644 --- a/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java +++ b/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java @@ -51,8 +51,12 @@ public final class MessageGenerator { static final String READABLE_CLASS = "org.apache.kafka.common.protocol.Readable"; + static final String RECORDS_READABLE_CLASS = "org.apache.kafka.common.protocol.RecordsReadable"; + static final String WRITABLE_CLASS = "org.apache.kafka.common.protocol.Writable"; + static final String RECORDS_WRITABLE_CLASS = "org.apache.kafka.common.protocol.RecordsWritable"; + static final String ARRAYS_CLASS = "java.util.Arrays"; static final String OBJECTS_CLASS = "java.util.Objects"; @@ -88,6 +92,10 @@ public final class MessageGenerator { static final String UUID_CLASS = "java.util.UUID"; + static final String BASE_RECORDS_CLASS = "org.apache.kafka.common.record.BaseRecords"; + + static final String MEMORY_RECORDS_CLASS = "org.apache.kafka.common.record.MemoryRecords"; + static final String REQUEST_SUFFIX = "Request"; static final String RESPONSE_SUFFIX = "Response"; diff --git a/generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java b/generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java index d433b0b01bef7..40075cfd8c5be 100644 --- a/generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java +++ b/generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java @@ -280,6 +280,8 @@ private String fieldTypeToSchemaType(FieldType type, } else { return nullable ? "Type.NULLABLE_BYTES" : "Type.BYTES"; } + } else if (type.isRecords()) { + return "Type.RECORDS"; } else if (type.isArray()) { if (fieldFlexibleVersions.contains(version)) { headerGenerator.addImport(MessageGenerator.COMPACT_ARRAYOF_CLASS); diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/common/FetchRequestBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/common/FetchRequestBenchmark.java new file mode 100644 index 0000000000000..5f68a89ef4971 --- /dev/null +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/common/FetchRequestBenchmark.java @@ -0,0 +1,129 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.kafka.jmh.common; + +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.network.Send; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.types.Struct; +import org.apache.kafka.common.requests.AbstractRequest; +import org.apache.kafka.common.requests.ByteBufferChannel; +import org.apache.kafka.common.requests.FetchRequest; +import org.apache.kafka.common.requests.RequestHeader; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Warmup; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.TimeUnit; + +@State(Scope.Benchmark) +@Fork(value = 1) +@Warmup(iterations = 5) +@Measurement(iterations = 15) +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +public class FetchRequestBenchmark { + + @Param({"10", "500", "1000"}) + private int topicCount; + + @Param({"3", "10", "20"}) + private int partitionCount; + + Map fetchData; + + RequestHeader header; + + FetchRequest consumerRequest; + + FetchRequest replicaRequest; + + Struct requestStruct; + + @Setup(Level.Trial) + public void setup() { + this.fetchData = new HashMap<>(); + for (int topicIdx = 0; topicIdx < topicCount; topicIdx++) { + String topic = UUID.randomUUID().toString(); + for (int partitionId = 0; partitionId < partitionCount; partitionId++) { + FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData( + 0, 0, 4096, Optional.empty()); + fetchData.put(new TopicPartition(topic, partitionId), partitionData); + } + } + + this.header = new RequestHeader(ApiKeys.FETCH, ApiKeys.FETCH.latestVersion(), "jmh-benchmark", 100); + this.consumerRequest = FetchRequest.Builder.forConsumer(0, 0, fetchData) + .build(ApiKeys.FETCH.latestVersion()); + this.replicaRequest = FetchRequest.Builder.forReplica(ApiKeys.FETCH.latestVersion(), 1, 0, 0, fetchData) + .build(ApiKeys.FETCH.latestVersion()); + this.requestStruct = this.consumerRequest.data().toStruct(ApiKeys.FETCH.latestVersion()); + + } + + @Benchmark + public short testFetchRequestFromStruct() { + AbstractRequest request = AbstractRequest.parseRequest(ApiKeys.FETCH, ApiKeys.FETCH.latestVersion(), requestStruct); + return request.version(); + } + + @Benchmark + public int testFetchRequestForConsumer() { + FetchRequest fetchRequest = FetchRequest.Builder.forConsumer(0, 0, fetchData) + .build(ApiKeys.FETCH.latestVersion()); + return fetchRequest.fetchData().size(); + } + + @Benchmark + public int testFetchRequestForReplica() { + FetchRequest fetchRequest = FetchRequest.Builder.forReplica( + ApiKeys.FETCH.latestVersion(), 1, 0, 0, fetchData) + .build(ApiKeys.FETCH.latestVersion()); + return fetchRequest.fetchData().size(); + } + + @Benchmark + public int testSerializeFetchRequestForConsumer() throws IOException { + Send send = consumerRequest.toSend("dest", header); + ByteBufferChannel channel = new ByteBufferChannel(send.size()); + send.writeTo(channel); + return channel.buffer().limit(); + } + + @Benchmark + public int testSerializeFetchRequestForReplica() throws IOException { + Send send = replicaRequest.toSend("dest", header); + ByteBufferChannel channel = new ByteBufferChannel(send.size()); + send.writeTo(channel); + return channel.buffer().limit(); + } +} diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/common/FetchResponseBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/common/FetchResponseBenchmark.java new file mode 100644 index 0000000000000..c742d67dbfecf --- /dev/null +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/common/FetchResponseBenchmark.java @@ -0,0 +1,104 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.kafka.jmh.common; + +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.network.Send; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.record.CompressionType; +import org.apache.kafka.common.record.MemoryRecords; +import org.apache.kafka.common.record.SimpleRecord; +import org.apache.kafka.common.requests.ByteBufferChannel; +import org.apache.kafka.common.requests.FetchResponse; +import org.apache.kafka.common.requests.ResponseHeader; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Warmup; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.TimeUnit; + +@State(Scope.Benchmark) +@Fork(value = 1) +@Warmup(iterations = 5) +@Measurement(iterations = 15) +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +public class FetchResponseBenchmark { + @Param({"10", "500", "1000"}) + private int topicCount; + + @Param({"3", "10", "20"}) + private int partitionCount; + + LinkedHashMap> responseData; + + ResponseHeader header; + + FetchResponse fetchResponse; + + @Setup(Level.Trial) + public void setup() { + MemoryRecords records = MemoryRecords.withRecords(CompressionType.NONE, + new SimpleRecord(1000, "key1".getBytes(StandardCharsets.UTF_8), "value1".getBytes(StandardCharsets.UTF_8)), + new SimpleRecord(1001, "key2".getBytes(StandardCharsets.UTF_8), "value2".getBytes(StandardCharsets.UTF_8)), + new SimpleRecord(1002, "key3".getBytes(StandardCharsets.UTF_8), "value3".getBytes(StandardCharsets.UTF_8))); + + this.responseData = new LinkedHashMap<>(); + for (int topicIdx = 0; topicIdx < topicCount; topicIdx++) { + String topic = UUID.randomUUID().toString(); + for (int partitionId = 0; partitionId < partitionCount; partitionId++) { + FetchResponse.PartitionData partitionData = new FetchResponse.PartitionData<>( + Errors.NONE, 0, 0, 0, Optional.empty(), Collections.emptyList(), records); + responseData.put(new TopicPartition(topic, partitionId), partitionData); + } + } + + this.header = new ResponseHeader(100, ApiKeys.FETCH.responseHeaderVersion(ApiKeys.FETCH.latestVersion())); + this.fetchResponse = new FetchResponse<>(Errors.NONE, responseData, 0, 0); + } + + @Benchmark + public int testConstructFetchResponse() { + FetchResponse fetchResponse = new FetchResponse<>(Errors.NONE, responseData, 0, 0); + return fetchResponse.responseData().size(); + } + + @Benchmark + public int testSerializeFetchResponse() throws IOException { + Send send = fetchResponse.toSend("dest", header, ApiKeys.FETCH.latestVersion()); + ByteBufferChannel channel = new ByteBufferChannel(send.size()); + send.writeTo(channel); + return channel.buffer().limit(); + } +}