You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2020/12/16 08:52:48 UTC
[GitHub] [kafka] chia7712 opened a new pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and duplicate construc…
chia7712 opened a new pull request #9758:
URL: https://github.com/apache/kafka/pull/9758
Replace stale class by auto-generated class.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r552047139
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -59,136 +60,140 @@
*/
public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
- public static final long INVALID_HIGHWATERMARK = -1L;
+ public static FetchResponseData.FetchablePartitionResponse partitionResponse(Errors error) {
+ return new FetchResponseData.FetchablePartitionResponse()
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK)
+ .setLastStableOffset(FetchResponse.INVALID_LAST_STABLE_OFFSET)
+ .setLogStartOffset(FetchResponse.INVALID_LOG_START_OFFSET)
+ .setAbortedTransactions(null)
+ .setRecordSet(MemoryRecords.EMPTY)
+ .setPreferredReadReplica(FetchResponse.INVALID_PREFERRED_REPLICA_ID);
+ }
+
+ public static final long INVALID_HIGH_WATERMARK = -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 FetchResponseData data;
- private final LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap;
+ private final LinkedHashMap<TopicPartition, FetchResponseData.FetchablePartitionResponse> partitionData;
+ // lazily instantiate this field
+ private LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap = null;
@Override
public FetchResponseData data() {
return data;
}
- public static final class AbortedTransaction {
- public final long producerId;
- public final long firstOffset;
+ /**
+ * From version 3 or later, the entries in `responseData` should be in the same order as the entries in
+ * `FetchRequest.fetchData`.
+ *
+ * @param error The top-level error code.
+ * @param responseData The fetched data grouped by partition.
+ * @param throttleTimeMs The time in milliseconds that the response was throttled
+ * @param sessionId The fetch session id.
+ */
+ public FetchResponse(Errors error,
+ LinkedHashMap<TopicPartition, PartitionData<T>> responseData,
+ int throttleTimeMs,
+ int sessionId) {
+ this(error, throttleTimeMs, sessionId, responseData.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey,
+ entry -> entry.getValue().partitionResponse, (o1, o2) -> {
+ throw new RuntimeException("this is impossible");
+ }, LinkedHashMap::new)));
+ }
- public AbortedTransaction(long producerId, long firstOffset) {
- this.producerId = producerId;
- this.firstOffset = firstOffset;
- }
+ public FetchResponse(Errors error,
+ int throttleTimeMs,
+ int sessionId,
+ LinkedHashMap<TopicPartition, FetchResponseData.FetchablePartitionResponse> responseData) {
+ super(ApiKeys.FETCH);
+ this.data = new FetchResponseData()
+ .setSessionId(sessionId)
+ .setErrorCode(error.code())
+ .setThrottleTimeMs(throttleTimeMs);
+ responseData.forEach((tp, tpData) -> data.responses().add(new FetchResponseData.FetchableTopicResponse()
+ .setTopic(tp.topic())
+ .setPartitionResponses(Collections.singletonList(tpData.setPartition(tp.partition())))));
+ this.partitionData = responseData;
+ }
- @Override
- public boolean equals(Object o) {
- if (this == o)
- return true;
- if (o == null || getClass() != o.getClass())
- return false;
+ public FetchResponse(FetchResponseData fetchResponseData) {
+ super(ApiKeys.FETCH);
+ this.data = fetchResponseData;
+ this.partitionData = new LinkedHashMap<>();
+ fetchResponseData.responses().forEach(topicResponse ->
+ topicResponse.partitionResponses().forEach(partitionResponse ->
+ partitionData.put(new TopicPartition(topicResponse.topic(), partitionResponse.partition()), partitionResponse))
+ );
+ }
- AbortedTransaction that = (AbortedTransaction) o;
+ public Errors error() {
+ return Errors.forCode(data.errorCode());
+ }
- return producerId == that.producerId && firstOffset == that.firstOffset;
+ public LinkedHashMap<TopicPartition, PartitionData<T>> responseData() {
+ if (responseDataMap == null) {
+ responseDataMap = new LinkedHashMap<>(partitionData.size());
+ partitionData.forEach((tp, d) -> responseDataMap.put(tp, new PartitionData<>(d)));
}
+ return responseDataMap;
+ }
- @Override
- public int hashCode() {
- int result = Long.hashCode(producerId);
- result = 31 * result + Long.hashCode(firstOffset);
- return result;
- }
+ @Override
+ public int throttleTimeMs() {
+ return data.throttleTimeMs();
+ }
- @Override
- public String toString() {
- return "(producerId=" + producerId + ", firstOffset=" + firstOffset + ")";
- }
+ public int sessionId() {
+ return data.sessionId();
+ }
- static AbortedTransaction fromMessage(FetchResponseData.AbortedTransaction abortedTransaction) {
- return new AbortedTransaction(abortedTransaction.producerId(), abortedTransaction.firstOffset());
- }
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ Map<Errors, Integer> errorCounts = new HashMap<>();
+ updateErrorCounts(errorCounts, error());
+ partitionData.values().forEach(response ->
+ updateErrorCounts(errorCounts, Errors.forCode(response.errorCode()))
+ );
+ return errorCounts;
}
- public static final class PartitionData<T extends BaseRecords> {
- private final FetchResponseData.FetchablePartitionResponse partitionResponse;
+ public static FetchResponse<MemoryRecords> parse(ByteBuffer buffer, short version) {
+ return new FetchResponse<>(new FetchResponseData(new ByteBufferAccessor(buffer), version));
+ }
- // Derived fields
- private final Optional<Integer> preferredReplica;
- private final List<AbortedTransaction> abortedTransactions;
- private final Errors error;
+ /**
+ * Convenience method to find the size of a response.
+ *
+ * @param version The version of the response to use.
+ * @param partIterator The partition iterator.
+ * @return The response size in bytes.
+ */
+ public static <T extends Records> int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ // Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
+ // use arbitrary values here without affecting the result.
+ LinkedHashMap<TopicPartition, PartitionData<T>> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
+ ObjectSerializationCache cache = new ObjectSerializationCache();
+ return 4 + new FetchResponse<>(Errors.NONE, data, 0, INVALID_SESSION_ID).data.size(cache, version);
+ }
- 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.preferredReadReplica())
- .filter(replicaId -> replicaId != INVALID_PREFERRED_REPLICA_ID);
-
- if (partitionResponse.abortedTransactions() == null) {
- this.abortedTransactions = null;
- } else {
- this.abortedTransactions = partitionResponse.abortedTransactions().stream()
- .map(AbortedTransaction::fromMessage)
- .collect(Collectors.toList());
- }
-
- this.error = Errors.forCode(partitionResponse.errorCode());
- }
+ @Override
+ public boolean shouldClientThrottle(short version) {
+ return version >= 8;
+ }
- public PartitionData(Errors error,
- long highWatermark,
- long lastStableOffset,
- long logStartOffset,
- Optional<Integer> preferredReadReplica,
- List<AbortedTransaction> abortedTransactions,
- Optional<FetchResponseData.EpochEndOffset> divergingEpoch,
- T records) {
- this.preferredReplica = preferredReadReplica;
- this.abortedTransactions = abortedTransactions;
- this.error = error;
-
- FetchResponseData.FetchablePartitionResponse partitionResponse =
- new FetchResponseData.FetchablePartitionResponse();
- partitionResponse.setErrorCode(error.code())
- .setHighWatermark(highWatermark)
- .setLastStableOffset(lastStableOffset)
- .setLogStartOffset(logStartOffset);
- if (abortedTransactions != null) {
- partitionResponse.setAbortedTransactions(abortedTransactions.stream().map(
- aborted -> new FetchResponseData.AbortedTransaction()
- .setProducerId(aborted.producerId)
- .setFirstOffset(aborted.firstOffset))
- .collect(Collectors.toList()));
- } else {
- partitionResponse.setAbortedTransactions(null);
- }
- partitionResponse.setPreferredReadReplica(preferredReadReplica.orElse(INVALID_PREFERRED_REPLICA_ID));
- partitionResponse.setRecordSet(records);
- divergingEpoch.ifPresent(partitionResponse::setDivergingEpoch);
- this.partitionResponse = partitionResponse;
- }
+ public static final class PartitionData<T extends BaseRecords> {
Review comment:
It seems like we do it for a couple of cases. It doesn't seem worth it given that fetch is one of the hot paths. We could have static helper methods instead if helpful.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-778804739
Sorry for the delay on this one. I'll try to review in the next few days if no-one else beats me to it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r585681834
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -761,79 +754,85 @@ class KafkaApis(val requestChannel: RequestChannel,
// For fetch requests from clients, check if down-conversion is disabled for the particular partition
if (!fetchRequest.isFromFollower && !logConfig.forall(_.messageDownConversionEnable)) {
trace(s"Conversion to message format ${downConvertMagic.get} is disabled for partition $tp. Sending unsupported version response to $clientId.")
- errorResponse(Errors.UNSUPPORTED_VERSION)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_VERSION)
} else {
try {
trace(s"Down converting records from partition $tp to message format version $magic for fetch request from $clientId")
// Because down-conversion is extremely memory intensive, we want to try and delay the down-conversion as much
// 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.
- val error = maybeDownConvertStorageError(partitionData.error)
- 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))
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(new LazyDownConversionRecords(tp, unconvertedRecords, magic, fetchContext.getFetchOffset(tp).get, time))
+ .setPreferredReadReplica(partitionData.preferredReadReplica())
} catch {
case e: UnsupportedCompressionTypeException =>
trace("Received unsupported compression type error during down-conversion", e)
- errorResponse(Errors.UNSUPPORTED_COMPRESSION_TYPE)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_COMPRESSION_TYPE)
}
}
case None =>
- val error = maybeDownConvertStorageError(partitionData.error)
- new FetchResponse.PartitionData[BaseRecords](error,
- partitionData.highWatermark,
- partitionData.lastStableOffset,
- partitionData.logStartOffset,
- partitionData.preferredReadReplica,
- partitionData.abortedTransactions,
- partitionData.divergingEpoch,
- unconvertedRecords)
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(unconvertedRecords)
+ .setPreferredReadReplica(partitionData.preferredReadReplica)
+ .setDivergingEpoch(partitionData.divergingEpoch)
}
}
}
// the callback for process a fetch response, invoked before throttling
def processResponseCallback(responsePartitionData: Seq[(TopicPartition, FetchPartitionData)]): Unit = {
- val partitions = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[Records]]
+ val partitions = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
val reassigningPartitions = mutable.Set[TopicPartition]()
responsePartitionData.foreach { case (tp, data) =>
val abortedTransactions = data.abortedTransactions.map(_.asJava).orNull
val lastStableOffset = data.lastStableOffset.getOrElse(FetchResponse.INVALID_LAST_STABLE_OFFSET)
- if (data.isReassignmentFetch)
- reassigningPartitions.add(tp)
- val error = maybeDownConvertStorageError(data.error)
- partitions.put(tp, new FetchResponse.PartitionData(
- error,
- data.highWatermark,
- lastStableOffset,
- data.logStartOffset,
- data.preferredReadReplica.map(int2Integer).asJava,
- abortedTransactions,
- data.divergingEpoch.asJava,
- data.records))
+ if (data.isReassignmentFetch) reassigningPartitions.add(tp)
+ val partitionData = new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(data.error).code)
+ .setHighWatermark(data.highWatermark)
+ .setLastStableOffset(lastStableOffset)
+ .setLogStartOffset(data.logStartOffset)
+ .setAbortedTransactions(abortedTransactions)
+ .setRecords(data.records)
+ .setPreferredReadReplica(data.preferredReadReplica.getOrElse(FetchResponse.INVALID_PREFERRED_REPLICA_ID))
+ data.divergingEpoch.foreach(partitionData.setDivergingEpoch)
+ partitions.put(tp, partitionData)
}
erroneous.foreach { case (tp, data) => partitions.put(tp, data) }
- var unconvertedFetchResponse: FetchResponse[Records] = null
+ var unconvertedFetchResponse: FetchResponse = null
- def createResponse(throttleTimeMs: Int): FetchResponse[BaseRecords] = {
+ def createResponse(throttleTimeMs: Int): FetchResponse = {
// Down-convert messages for each partition if required
- val convertedData = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[BaseRecords]]
+ val convertedData = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
unconvertedFetchResponse.responseData.forEach { (tp, unconvertedPartitionData) =>
- if (unconvertedPartitionData.error != Errors.NONE)
+ val error = Errors.forCode(unconvertedPartitionData.errorCode)
+ if (error != Errors.NONE)
debug(s"Fetch request with correlation id ${request.header.correlationId} from client $clientId " +
- s"on partition $tp failed due to ${unconvertedPartitionData.error.exceptionName}")
+ s"on partition $tp failed due to ${error.exceptionName}")
convertedData.put(tp, maybeConvertFetchedData(tp, unconvertedPartitionData))
}
// Prepare fetch response from converted data
- val response = new FetchResponse(unconvertedFetchResponse.error, convertedData, throttleTimeMs,
- unconvertedFetchResponse.sessionId)
+ val response = FetchResponse.of(unconvertedFetchResponse.error, throttleTimeMs, unconvertedFetchResponse.sessionId, convertedData)
// record the bytes out metrics only when the response is being sent
response.responseData.forEach { (tp, data) =>
- brokerTopicStats.updateBytesOut(tp.topic, fetchRequest.isFromFollower, reassigningPartitions.contains(tp), data.records.sizeInBytes)
+ brokerTopicStats.updateBytesOut(tp.topic, fetchRequest.isFromFollower,
+ reassigningPartitions.contains(tp), if (data.records == null) 0 else data.records.sizeInBytes)
Review comment:
The converted record is `LazyDownConversionRecords` so casting it to `Records` can produce error.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r585691505
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -761,79 +754,85 @@ class KafkaApis(val requestChannel: RequestChannel,
// For fetch requests from clients, check if down-conversion is disabled for the particular partition
if (!fetchRequest.isFromFollower && !logConfig.forall(_.messageDownConversionEnable)) {
trace(s"Conversion to message format ${downConvertMagic.get} is disabled for partition $tp. Sending unsupported version response to $clientId.")
- errorResponse(Errors.UNSUPPORTED_VERSION)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_VERSION)
} else {
try {
trace(s"Down converting records from partition $tp to message format version $magic for fetch request from $clientId")
// Because down-conversion is extremely memory intensive, we want to try and delay the down-conversion as much
// 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.
- val error = maybeDownConvertStorageError(partitionData.error)
- 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))
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(new LazyDownConversionRecords(tp, unconvertedRecords, magic, fetchContext.getFetchOffset(tp).get, time))
+ .setPreferredReadReplica(partitionData.preferredReadReplica())
} catch {
case e: UnsupportedCompressionTypeException =>
trace("Received unsupported compression type error during down-conversion", e)
- errorResponse(Errors.UNSUPPORTED_COMPRESSION_TYPE)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_COMPRESSION_TYPE)
}
}
case None =>
- val error = maybeDownConvertStorageError(partitionData.error)
- new FetchResponse.PartitionData[BaseRecords](error,
- partitionData.highWatermark,
- partitionData.lastStableOffset,
- partitionData.logStartOffset,
- partitionData.preferredReadReplica,
- partitionData.abortedTransactions,
- partitionData.divergingEpoch,
- unconvertedRecords)
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(unconvertedRecords)
+ .setPreferredReadReplica(partitionData.preferredReadReplica)
+ .setDivergingEpoch(partitionData.divergingEpoch)
}
}
}
// the callback for process a fetch response, invoked before throttling
def processResponseCallback(responsePartitionData: Seq[(TopicPartition, FetchPartitionData)]): Unit = {
- val partitions = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[Records]]
+ val partitions = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
val reassigningPartitions = mutable.Set[TopicPartition]()
responsePartitionData.foreach { case (tp, data) =>
val abortedTransactions = data.abortedTransactions.map(_.asJava).orNull
val lastStableOffset = data.lastStableOffset.getOrElse(FetchResponse.INVALID_LAST_STABLE_OFFSET)
- if (data.isReassignmentFetch)
- reassigningPartitions.add(tp)
- val error = maybeDownConvertStorageError(data.error)
- partitions.put(tp, new FetchResponse.PartitionData(
- error,
- data.highWatermark,
- lastStableOffset,
- data.logStartOffset,
- data.preferredReadReplica.map(int2Integer).asJava,
- abortedTransactions,
- data.divergingEpoch.asJava,
- data.records))
+ if (data.isReassignmentFetch) reassigningPartitions.add(tp)
+ val partitionData = new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(data.error).code)
+ .setHighWatermark(data.highWatermark)
+ .setLastStableOffset(lastStableOffset)
+ .setLogStartOffset(data.logStartOffset)
+ .setAbortedTransactions(abortedTransactions)
+ .setRecords(data.records)
+ .setPreferredReadReplica(data.preferredReadReplica.getOrElse(FetchResponse.INVALID_PREFERRED_REPLICA_ID))
+ data.divergingEpoch.foreach(partitionData.setDivergingEpoch)
+ partitions.put(tp, partitionData)
}
erroneous.foreach { case (tp, data) => partitions.put(tp, data) }
- var unconvertedFetchResponse: FetchResponse[Records] = null
+ var unconvertedFetchResponse: FetchResponse = null
- def createResponse(throttleTimeMs: Int): FetchResponse[BaseRecords] = {
+ def createResponse(throttleTimeMs: Int): FetchResponse = {
// Down-convert messages for each partition if required
- val convertedData = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[BaseRecords]]
+ val convertedData = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
unconvertedFetchResponse.responseData.forEach { (tp, unconvertedPartitionData) =>
- if (unconvertedPartitionData.error != Errors.NONE)
+ val error = Errors.forCode(unconvertedPartitionData.errorCode)
+ if (error != Errors.NONE)
debug(s"Fetch request with correlation id ${request.header.correlationId} from client $clientId " +
- s"on partition $tp failed due to ${unconvertedPartitionData.error.exceptionName}")
+ s"on partition $tp failed due to ${error.exceptionName}")
convertedData.put(tp, maybeConvertFetchedData(tp, unconvertedPartitionData))
}
// Prepare fetch response from converted data
- val response = new FetchResponse(unconvertedFetchResponse.error, convertedData, throttleTimeMs,
- unconvertedFetchResponse.sessionId)
+ val response = FetchResponse.of(unconvertedFetchResponse.error, throttleTimeMs, unconvertedFetchResponse.sessionId, convertedData)
// record the bytes out metrics only when the response is being sent
response.responseData.forEach { (tp, data) =>
- brokerTopicStats.updateBytesOut(tp.topic, fetchRequest.isFromFollower, reassigningPartitions.contains(tp), data.records.sizeInBytes)
+ brokerTopicStats.updateBytesOut(tp.topic, fetchRequest.isFromFollower,
+ reassigningPartitions.contains(tp), if (data.records == null) 0 else data.records.sizeInBytes)
Review comment:
Ah, I thought the `size` method was in `FetchResponse`. Maybe we should add one (`recordsSize` or something)? That would always be safe to call.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r585219789
##########
File path: core/src/main/scala/kafka/server/AbstractFetcherThread.scala
##########
@@ -416,9 +412,8 @@ abstract class AbstractFetcherThread(name: String,
"expected to persist.")
partitionsWithError += topicPartition
- case _ =>
- error(s"Error for partition $topicPartition at offset ${currentFetchState.fetchOffset}",
- partitionData.error.exception)
+ case partitionError: Errors =>
Review comment:
you are right. fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584224855
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -761,76 +754,80 @@ class KafkaApis(val requestChannel: RequestChannel,
// For fetch requests from clients, check if down-conversion is disabled for the particular partition
if (!fetchRequest.isFromFollower && !logConfig.forall(_.messageDownConversionEnable)) {
trace(s"Conversion to message format ${downConvertMagic.get} is disabled for partition $tp. Sending unsupported version response to $clientId.")
- errorResponse(Errors.UNSUPPORTED_VERSION)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_VERSION)
} else {
try {
trace(s"Down converting records from partition $tp to message format version $magic for fetch request from $clientId")
// Because down-conversion is extremely memory intensive, we want to try and delay the down-conversion as much
// 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.
- val error = maybeDownConvertStorageError(partitionData.error)
- 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))
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(new LazyDownConversionRecords(tp, unconvertedRecords, magic, fetchContext.getFetchOffset(tp).get, time))
+ .setPreferredReadReplica(partitionData.preferredReadReplica())
} catch {
case e: UnsupportedCompressionTypeException =>
trace("Received unsupported compression type error during down-conversion", e)
- errorResponse(Errors.UNSUPPORTED_COMPRESSION_TYPE)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_COMPRESSION_TYPE)
}
}
case None =>
- val error = maybeDownConvertStorageError(partitionData.error)
- new FetchResponse.PartitionData[BaseRecords](error,
- partitionData.highWatermark,
- partitionData.lastStableOffset,
- partitionData.logStartOffset,
- partitionData.preferredReadReplica,
- partitionData.abortedTransactions,
- partitionData.divergingEpoch,
- unconvertedRecords)
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(unconvertedRecords)
+ .setPreferredReadReplica(partitionData.preferredReadReplica)
+ .setDivergingEpoch(partitionData.divergingEpoch)
}
}
}
// the callback for process a fetch response, invoked before throttling
def processResponseCallback(responsePartitionData: Seq[(TopicPartition, FetchPartitionData)]): Unit = {
- val partitions = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[Records]]
+ val partitions = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
val reassigningPartitions = mutable.Set[TopicPartition]()
responsePartitionData.foreach { case (tp, data) =>
val abortedTransactions = data.abortedTransactions.map(_.asJava).orNull
val lastStableOffset = data.lastStableOffset.getOrElse(FetchResponse.INVALID_LAST_STABLE_OFFSET)
- if (data.isReassignmentFetch)
- reassigningPartitions.add(tp)
- val error = maybeDownConvertStorageError(data.error)
- partitions.put(tp, new FetchResponse.PartitionData(
- error,
- data.highWatermark,
- lastStableOffset,
- data.logStartOffset,
- data.preferredReadReplica.map(int2Integer).asJava,
- abortedTransactions,
- data.divergingEpoch.asJava,
- data.records))
+ if (data.isReassignmentFetch) reassigningPartitions.add(tp)
+ partitions.put(tp, new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(data.error).code)
+ .setHighWatermark(data.highWatermark)
+ .setLastStableOffset(lastStableOffset)
+ .setLogStartOffset(data.logStartOffset)
+ .setAbortedTransactions(abortedTransactions)
+ .setRecords(data.records)
+ .setPreferredReadReplica(data.preferredReadReplica.getOrElse(FetchResponse.INVALID_PREFERRED_REPLICA_ID))
+ .setDivergingEpoch(data.divergingEpoch.getOrElse(new FetchResponseData.EpochEndOffset)))
Review comment:
same to #9758 (comment)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 merged pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 merged pull request #9758:
URL: https://github.com/apache/kafka/pull/9758
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r552606639
##########
File path: core/src/main/scala/kafka/server/FetchSession.scala
##########
@@ -142,19 +141,20 @@ class CachedPartition(val topic: String,
if (updateResponseData)
localLogStartOffset = respData.logStartOffset
}
- if (respData.preferredReadReplica.isPresent) {
+ if (FetchResponse.preferredReadReplica(respData).isPresent) {
Review comment:
Maybe we could have a isPreferredReplica method since there are at least two places where we just call isPresent on the result.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-793348223
> I noticed that all of these JIRAs are listed as open, but they reference this PR. Are there further changes to be made? Asking because I'm working on similar code for #9944 and want to minimize merge conflicts as best as possible.
thanks for this response. Only #10269 is in progress. I'm sorry that this PR brings a bunch of conflicting files to #9944. I'd like to merge #9944 before other minor refactor-related patches.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r585224829
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -761,79 +754,84 @@ class KafkaApis(val requestChannel: RequestChannel,
// For fetch requests from clients, check if down-conversion is disabled for the particular partition
if (!fetchRequest.isFromFollower && !logConfig.forall(_.messageDownConversionEnable)) {
trace(s"Conversion to message format ${downConvertMagic.get} is disabled for partition $tp. Sending unsupported version response to $clientId.")
- errorResponse(Errors.UNSUPPORTED_VERSION)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_VERSION)
} else {
try {
trace(s"Down converting records from partition $tp to message format version $magic for fetch request from $clientId")
// Because down-conversion is extremely memory intensive, we want to try and delay the down-conversion as much
// 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.
- val error = maybeDownConvertStorageError(partitionData.error)
- 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))
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(new LazyDownConversionRecords(tp, unconvertedRecords, magic, fetchContext.getFetchOffset(tp).get, time))
+ .setPreferredReadReplica(partitionData.preferredReadReplica())
} catch {
case e: UnsupportedCompressionTypeException =>
trace("Received unsupported compression type error during down-conversion", e)
- errorResponse(Errors.UNSUPPORTED_COMPRESSION_TYPE)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_COMPRESSION_TYPE)
}
}
case None =>
- val error = maybeDownConvertStorageError(partitionData.error)
- new FetchResponse.PartitionData[BaseRecords](error,
- partitionData.highWatermark,
- partitionData.lastStableOffset,
- partitionData.logStartOffset,
- partitionData.preferredReadReplica,
- partitionData.abortedTransactions,
- partitionData.divergingEpoch,
- unconvertedRecords)
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(unconvertedRecords)
+ .setPreferredReadReplica(partitionData.preferredReadReplica)
+ .setDivergingEpoch(partitionData.divergingEpoch)
}
}
}
// the callback for process a fetch response, invoked before throttling
def processResponseCallback(responsePartitionData: Seq[(TopicPartition, FetchPartitionData)]): Unit = {
- val partitions = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[Records]]
+ val partitions = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
val reassigningPartitions = mutable.Set[TopicPartition]()
responsePartitionData.foreach { case (tp, data) =>
val abortedTransactions = data.abortedTransactions.map(_.asJava).orNull
val lastStableOffset = data.lastStableOffset.getOrElse(FetchResponse.INVALID_LAST_STABLE_OFFSET)
- if (data.isReassignmentFetch)
- reassigningPartitions.add(tp)
- val error = maybeDownConvertStorageError(data.error)
- partitions.put(tp, new FetchResponse.PartitionData(
- error,
- data.highWatermark,
- lastStableOffset,
- data.logStartOffset,
- data.preferredReadReplica.map(int2Integer).asJava,
- abortedTransactions,
- data.divergingEpoch.asJava,
- data.records))
+ if (data.isReassignmentFetch) reassigningPartitions.add(tp)
+ partitions.put(tp, new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(data.error).code)
+ .setHighWatermark(data.highWatermark)
+ .setLastStableOffset(lastStableOffset)
+ .setLogStartOffset(data.logStartOffset)
+ .setAbortedTransactions(abortedTransactions)
+ .setRecords(data.records)
+ .setPreferredReadReplica(data.preferredReadReplica.getOrElse(FetchResponse.INVALID_PREFERRED_REPLICA_ID))
+ .setDivergingEpoch(data.divergingEpoch.getOrElse(new FetchResponseData.EpochEndOffset)))
Review comment:
make sense. fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r585674420
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -365,17 +126,98 @@ public int sessionId() {
* @param partIterator The partition iterator.
* @return The response size in bytes.
*/
- public static <T extends BaseRecords> int sizeOf(short version,
- Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ public static int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, FetchResponseData.PartitionData>> partIterator) {
// Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
// use arbitrary values here without affecting the result.
- FetchResponseData data = toMessage(0, Errors.NONE, partIterator, INVALID_SESSION_ID);
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
ObjectSerializationCache cache = new ObjectSerializationCache();
- return 4 + data.size(cache, version);
+ return 4 + FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, data).data.size(cache, version);
}
@Override
public boolean shouldClientThrottle(short version) {
return version >= 8;
}
-}
+
+ public static Optional<FetchResponseData.EpochEndOffset> divergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() < 0 ? Optional.empty()
+ : Optional.of(partitionResponse.divergingEpoch());
+ }
+
+ public static boolean isDivergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() >= 0;
+ }
+
+ public static Optional<Integer> preferredReadReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() == INVALID_PREFERRED_REPLICA_ID ? Optional.empty()
+ : Optional.of(partitionResponse.preferredReadReplica());
+ }
+
+ public static boolean isPreferredReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() != INVALID_PREFERRED_REPLICA_ID;
+ }
+
+ public static FetchResponseData.PartitionData partitionResponse(int partition, Errors error) {
+ return new FetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK);
+ }
+
+ /**
+ * Returns `partition.records` as `Records` (instead of `BaseRecords`). If `records` is `null`, returns `MemoryRecords.EMPTY`.
+ *
+ * If this response was deserialized after a fetch, this method should never fail. An example where this would
+ * fail is a down-converted response (e.g. LazyDownConversionRecords) on the broker (before it's serialized and
+ * sent on the wire).
+ *
+ * @param partition partition data
+ * @return Records or empty record if the records in PartitionData is null.
+ */
+ public static Records records(FetchResponseData.PartitionData partition) {
+ if (partition.records() == null) return MemoryRecords.EMPTY;
+ else if (partition.records() instanceof Records) return (Records) partition.records();
+ else throw new IllegalStateException("the record type is " + partition.records().getClass().getSimpleName() +
Review comment:
Will address those nice comments
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584142074
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -57,238 +56,39 @@
* the fetch offset after the index lookup
* - {@link Errors#UNKNOWN_SERVER_ERROR} For any unexpected errors
*/
-public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
-
- public static final long INVALID_HIGHWATERMARK = -1L;
+public class FetchResponse extends AbstractResponse {
+ public static final long INVALID_HIGH_WATERMARK = -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 FetchResponseData data;
- private final LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap;
+ // we build responseData when needed.
+ private LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData = null;
Review comment:
Looks like this method is used only `Fetcher.sendFetches` outside of tests, benchmaks, etc.. Also once in `errorCounts`, but we can change that code. Maybe we can remove this altogether. Thoughts?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jolshan commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
jolshan commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-794336035
@chia7712 I see. Thanks for the response. I'm not sure if https://github.com/apache/kafka/pull/9944 will be ready after fixing conflicts. There were still a few critical tests failing. I can try to fix those but still figuring out the timeline.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-794327880
> So should we go forward with #10269 first still?
It needs some time to complete performance tests. It seems to me we can fix conflicting files in #9944 to get reviews and then push it before merging #10269. BTW, don't worry about causing potential conflicting files to #10269 :)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r581069071
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -57,238 +55,51 @@
* the fetch offset after the index lookup
* - {@link Errors#UNKNOWN_SERVER_ERROR} For any unexpected errors
*/
-public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
-
- public static final long INVALID_HIGHWATERMARK = -1L;
+public class FetchResponse extends AbstractResponse {
+ public static final long INVALID_HIGH_WATERMARK = -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 FetchResponseData data;
- private final LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap;
+ private final LinkedHashMap<TopicPartition, FetchResponseData.FetchablePartitionResponse> responseData;
@Override
public FetchResponseData data() {
return data;
}
- public static final class AbortedTransaction {
- public final long producerId;
- public final long firstOffset;
-
- public AbortedTransaction(long producerId, long firstOffset) {
- this.producerId = producerId;
- this.firstOffset = firstOffset;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o)
- return true;
- if (o == null || getClass() != o.getClass())
- return false;
-
- AbortedTransaction that = (AbortedTransaction) o;
-
- return producerId == that.producerId && firstOffset == that.firstOffset;
- }
-
- @Override
- public int hashCode() {
- int result = Long.hashCode(producerId);
- result = 31 * result + Long.hashCode(firstOffset);
- return result;
- }
-
- @Override
- 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<T extends BaseRecords> {
- private final FetchResponseData.FetchablePartitionResponse partitionResponse;
-
- // Derived fields
- private final Optional<Integer> preferredReplica;
- private final List<AbortedTransaction> 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.preferredReadReplica())
- .filter(replicaId -> replicaId != INVALID_PREFERRED_REPLICA_ID);
-
- if (partitionResponse.abortedTransactions() == null) {
- this.abortedTransactions = null;
- } else {
- this.abortedTransactions = partitionResponse.abortedTransactions().stream()
- .map(AbortedTransaction::fromMessage)
- .collect(Collectors.toList());
- }
-
- this.error = Errors.forCode(partitionResponse.errorCode());
- }
-
- public PartitionData(Errors error,
- long highWatermark,
- long lastStableOffset,
- long logStartOffset,
- Optional<Integer> preferredReadReplica,
- List<AbortedTransaction> abortedTransactions,
- Optional<FetchResponseData.EpochEndOffset> divergingEpoch,
- T records) {
- this.preferredReplica = preferredReadReplica;
- this.abortedTransactions = abortedTransactions;
- this.error = error;
-
- FetchResponseData.FetchablePartitionResponse partitionResponse =
- new FetchResponseData.FetchablePartitionResponse();
- partitionResponse.setErrorCode(error.code())
- .setHighWatermark(highWatermark)
- .setLastStableOffset(lastStableOffset)
- .setLogStartOffset(logStartOffset);
- if (abortedTransactions != null) {
- partitionResponse.setAbortedTransactions(abortedTransactions.stream().map(
- aborted -> new FetchResponseData.AbortedTransaction()
- .setProducerId(aborted.producerId)
- .setFirstOffset(aborted.firstOffset))
- .collect(Collectors.toList()));
- } else {
- partitionResponse.setAbortedTransactions(null);
- }
- partitionResponse.setPreferredReadReplica(preferredReadReplica.orElse(INVALID_PREFERRED_REPLICA_ID));
- partitionResponse.setRecordSet(records);
- divergingEpoch.ifPresent(partitionResponse::setDivergingEpoch);
-
- this.partitionResponse = partitionResponse;
- }
-
- public PartitionData(Errors error,
- long highWatermark,
- long lastStableOffset,
- long logStartOffset,
- Optional<Integer> preferredReadReplica,
- List<AbortedTransaction> abortedTransactions,
- T records) {
- this(error, highWatermark, lastStableOffset, logStartOffset, preferredReadReplica,
- abortedTransactions, Optional.empty(), records);
- }
-
- public PartitionData(Errors error,
- long highWatermark,
- long lastStableOffset,
- long logStartOffset,
- List<AbortedTransaction> abortedTransactions,
- T records) {
- this(error, highWatermark, lastStableOffset, logStartOffset, Optional.empty(), abortedTransactions, records);
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o)
- return true;
- if (o == null || getClass() != o.getClass())
- return false;
-
- PartitionData that = (PartitionData) o;
-
- return this.partitionResponse.equals(that.partitionResponse);
- }
-
- @Override
- public int hashCode() {
- 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() +
- ", divergingEpoch =" + divergingEpoch() +
- ", recordsSizeInBytes=" + records().sizeInBytes() + ")";
- }
-
- public Errors error() {
- return error;
- }
-
- public long highWatermark() {
- return partitionResponse.highWatermark();
- }
-
- public long lastStableOffset() {
- return partitionResponse.lastStableOffset();
- }
-
- public long logStartOffset() {
- return partitionResponse.logStartOffset();
- }
-
- public Optional<Integer> preferredReadReplica() {
- return preferredReplica;
- }
-
- public List<AbortedTransaction> abortedTransactions() {
- return abortedTransactions;
- }
-
- public Optional<FetchResponseData.EpochEndOffset> divergingEpoch() {
- FetchResponseData.EpochEndOffset epochEndOffset = partitionResponse.divergingEpoch();
- if (epochEndOffset.epoch() < 0) {
- return Optional.empty();
- } else {
- return Optional.of(epochEndOffset);
- }
- }
-
- @SuppressWarnings("unchecked")
- public T records() {
- return (T) partitionResponse.recordSet();
- }
- }
-
- /**
- * From version 3 or later, the entries in `responseData` should be in the same order as the entries in
- * `FetchRequest.fetchData`.
- *
- * @param error The top-level error code.
- * @param responseData The fetched data grouped by partition.
- * @param throttleTimeMs The time in milliseconds that the response was throttled
- * @param sessionId The fetch session id.
- */
public FetchResponse(Errors error,
- LinkedHashMap<TopicPartition, PartitionData<T>> responseData,
int throttleTimeMs,
- int sessionId) {
- super(ApiKeys.FETCH);
- this.data = toMessage(throttleTimeMs, error, responseData.entrySet().iterator(), sessionId);
- this.responseDataMap = responseData;
+ int sessionId,
+ LinkedHashMap<TopicPartition, FetchResponseData.FetchablePartitionResponse> responseData) {
+ this(new FetchResponseData()
+ .setSessionId(sessionId)
+ .setErrorCode(error.code())
+ .setThrottleTimeMs(throttleTimeMs)
+ .setResponses(responseData.entrySet().stream().map(entry -> new FetchResponseData.FetchableTopicResponse()
+ .setTopic(entry.getKey().topic())
+ .setPartitionResponses(Collections.singletonList(entry.getValue().setPartition(entry.getKey().partition()))))
+ .collect(Collectors.toList())));
}
public FetchResponse(FetchResponseData fetchResponseData) {
super(ApiKeys.FETCH);
this.data = fetchResponseData;
- this.responseDataMap = toResponseDataMap(fetchResponseData);
+ this.responseData = new LinkedHashMap<>();
Review comment:
This isn't needed when we return a fetch from the broker, right? If this is true, can we remove it from the fetch response and build it on the client when needed?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r585226663
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -365,17 +126,92 @@ public int sessionId() {
* @param partIterator The partition iterator.
* @return The response size in bytes.
*/
- public static <T extends BaseRecords> int sizeOf(short version,
- Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ public static int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, FetchResponseData.PartitionData>> partIterator) {
// Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
// use arbitrary values here without affecting the result.
- FetchResponseData data = toMessage(0, Errors.NONE, partIterator, INVALID_SESSION_ID);
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
ObjectSerializationCache cache = new ObjectSerializationCache();
- return 4 + data.size(cache, version);
+ return 4 + FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, data).data.size(cache, version);
}
@Override
public boolean shouldClientThrottle(short version) {
return version >= 8;
}
-}
+
+ public static Optional<FetchResponseData.EpochEndOffset> divergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() < 0 ? Optional.empty()
+ : Optional.of(partitionResponse.divergingEpoch());
+ }
+
+ public static boolean isDivergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() >= 0;
+ }
+
+ public static Optional<Integer> preferredReadReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() == INVALID_PREFERRED_REPLICA_ID ? Optional.empty()
+ : Optional.of(partitionResponse.preferredReadReplica());
+ }
+
+ public static boolean isPreferredReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() != INVALID_PREFERRED_REPLICA_ID;
+ }
+
+ public static FetchResponseData.PartitionData partitionResponse(int partition, Errors error) {
+ return new FetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK);
+ }
+
+ /**
+ * cast the BaseRecords of PartitionData to Records. KRPC converts the byte array to MemoryRecords so this method
+ * never fail if the data is from KRPC.
Review comment:
good one. will copy that
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-790614014
Great to see this merged. @chia7712 Will you follow up with the items we split into separate JIRAs? I think the 3 key ones are:
1. Remove unnecessary copying, if possible, in KafkaApis
2. Remove the lazily populated map in FetchResponse
3. I think you had an idea for removing `FetchResponse.toMessage`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584224981
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -761,76 +754,80 @@ class KafkaApis(val requestChannel: RequestChannel,
// For fetch requests from clients, check if down-conversion is disabled for the particular partition
if (!fetchRequest.isFromFollower && !logConfig.forall(_.messageDownConversionEnable)) {
trace(s"Conversion to message format ${downConvertMagic.get} is disabled for partition $tp. Sending unsupported version response to $clientId.")
- errorResponse(Errors.UNSUPPORTED_VERSION)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_VERSION)
} else {
try {
trace(s"Down converting records from partition $tp to message format version $magic for fetch request from $clientId")
// Because down-conversion is extremely memory intensive, we want to try and delay the down-conversion as much
// 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.
- val error = maybeDownConvertStorageError(partitionData.error)
- 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))
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(new LazyDownConversionRecords(tp, unconvertedRecords, magic, fetchContext.getFetchOffset(tp).get, time))
+ .setPreferredReadReplica(partitionData.preferredReadReplica())
Review comment:
https://issues.apache.org/jira/browse/KAFKA-12387
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r581044552
##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
##########
@@ -290,7 +291,7 @@ public void onSuccess(ClientResponse resp) {
Set<TopicPartition> partitions = new HashSet<>(response.responseData().keySet());
FetchResponseMetricAggregator metricAggregator = new FetchResponseMetricAggregator(sensors, partitions);
- for (Map.Entry<TopicPartition, FetchResponse.PartitionData<Records>> entry : response.responseData().entrySet()) {
+ for (Map.Entry<TopicPartition, FetchResponseData.FetchablePartitionResponse> entry : response.responseData().entrySet()) {
Review comment:
`FetchablePartitionResponse` is a bit long and redundant. Could we find a shorter name for it now that `PartitionData` is gone?
##########
File path: jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetchsession/FetchSessionBenchmark.java
##########
@@ -70,24 +70,26 @@ public void setUp() {
handler = new FetchSessionHandler(LOG_CONTEXT, 1);
FetchSessionHandler.Builder builder = handler.newBuilder();
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> respMap = new LinkedHashMap<>();
+ LinkedHashMap<TopicPartition, FetchResponseData.FetchablePartitionResponse> respMap = new LinkedHashMap<>();
for (int i = 0; i < partitionCount; i++) {
TopicPartition tp = new TopicPartition("foo", i);
FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData(0, 0, 200,
Optional.empty());
fetches.put(tp, partitionData);
builder.add(tp, partitionData);
- respMap.put(tp, new FetchResponse.PartitionData<>(
- Errors.NONE,
- 0L,
- 0L,
- 0,
- null,
- null));
+ respMap.put(tp, new FetchResponseData.FetchablePartitionResponse()
+ .setPartition(tp.partition())
+ .setErrorCode(Errors.NONE.code())
+ .setHighWatermark(0)
+ .setLastStableOffset(0)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(null)
+ .setRecords(null)
+ .setPreferredReadReplica(FetchResponse.INVALID_PREFERRED_REPLICA_ID));
Review comment:
Maybe we can remove the defaults from this and every other place where we build `FetchablePartitionResponse`
##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
##########
@@ -1257,7 +1259,7 @@ private CompletedFetch initializeCompletedFetch(CompletedFetch nextCompletedFetc
log.trace("Preparing to read {} bytes of data for partition {} with offset {}",
partition.records().sizeInBytes(), tp, position);
- Iterator<? extends RecordBatch> batches = partition.records().batches().iterator();
+ Iterator<? extends RecordBatch> batches = ((Records) partition.records()).batches().iterator();
Review comment:
Could we encapsulate this cast in a utility method?
##########
File path: core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala
##########
@@ -110,7 +113,7 @@ class ReplicaAlterLogDirsThread(name: String,
// process fetched data
override def processPartitionData(topicPartition: TopicPartition,
fetchOffset: Long,
- partitionData: PartitionData[Records]): Option[LogAppendInfo] = {
+ partitionData: FetchResponseData.FetchablePartitionResponse): Option[LogAppendInfo] = {
Review comment:
Could this be `FetchData` too? Are there other places like it?
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -57,238 +55,51 @@
* the fetch offset after the index lookup
* - {@link Errors#UNKNOWN_SERVER_ERROR} For any unexpected errors
*/
-public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
-
- public static final long INVALID_HIGHWATERMARK = -1L;
+public class FetchResponse extends AbstractResponse {
+ public static final long INVALID_HIGH_WATERMARK = -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 FetchResponseData data;
- private final LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap;
+ private final LinkedHashMap<TopicPartition, FetchResponseData.FetchablePartitionResponse> responseData;
@Override
public FetchResponseData data() {
return data;
}
- public static final class AbortedTransaction {
- public final long producerId;
- public final long firstOffset;
-
- public AbortedTransaction(long producerId, long firstOffset) {
- this.producerId = producerId;
- this.firstOffset = firstOffset;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o)
- return true;
- if (o == null || getClass() != o.getClass())
- return false;
-
- AbortedTransaction that = (AbortedTransaction) o;
-
- return producerId == that.producerId && firstOffset == that.firstOffset;
- }
-
- @Override
- public int hashCode() {
- int result = Long.hashCode(producerId);
- result = 31 * result + Long.hashCode(firstOffset);
- return result;
- }
-
- @Override
- 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<T extends BaseRecords> {
- private final FetchResponseData.FetchablePartitionResponse partitionResponse;
-
- // Derived fields
- private final Optional<Integer> preferredReplica;
- private final List<AbortedTransaction> 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.preferredReadReplica())
- .filter(replicaId -> replicaId != INVALID_PREFERRED_REPLICA_ID);
-
- if (partitionResponse.abortedTransactions() == null) {
- this.abortedTransactions = null;
- } else {
- this.abortedTransactions = partitionResponse.abortedTransactions().stream()
- .map(AbortedTransaction::fromMessage)
- .collect(Collectors.toList());
- }
-
- this.error = Errors.forCode(partitionResponse.errorCode());
- }
-
- public PartitionData(Errors error,
- long highWatermark,
- long lastStableOffset,
- long logStartOffset,
- Optional<Integer> preferredReadReplica,
- List<AbortedTransaction> abortedTransactions,
- Optional<FetchResponseData.EpochEndOffset> divergingEpoch,
- T records) {
- this.preferredReplica = preferredReadReplica;
- this.abortedTransactions = abortedTransactions;
- this.error = error;
-
- FetchResponseData.FetchablePartitionResponse partitionResponse =
- new FetchResponseData.FetchablePartitionResponse();
- partitionResponse.setErrorCode(error.code())
- .setHighWatermark(highWatermark)
- .setLastStableOffset(lastStableOffset)
- .setLogStartOffset(logStartOffset);
- if (abortedTransactions != null) {
- partitionResponse.setAbortedTransactions(abortedTransactions.stream().map(
- aborted -> new FetchResponseData.AbortedTransaction()
- .setProducerId(aborted.producerId)
- .setFirstOffset(aborted.firstOffset))
- .collect(Collectors.toList()));
- } else {
- partitionResponse.setAbortedTransactions(null);
- }
- partitionResponse.setPreferredReadReplica(preferredReadReplica.orElse(INVALID_PREFERRED_REPLICA_ID));
- partitionResponse.setRecordSet(records);
- divergingEpoch.ifPresent(partitionResponse::setDivergingEpoch);
-
- this.partitionResponse = partitionResponse;
- }
-
- public PartitionData(Errors error,
- long highWatermark,
- long lastStableOffset,
- long logStartOffset,
- Optional<Integer> preferredReadReplica,
- List<AbortedTransaction> abortedTransactions,
- T records) {
- this(error, highWatermark, lastStableOffset, logStartOffset, preferredReadReplica,
- abortedTransactions, Optional.empty(), records);
- }
-
- public PartitionData(Errors error,
- long highWatermark,
- long lastStableOffset,
- long logStartOffset,
- List<AbortedTransaction> abortedTransactions,
- T records) {
- this(error, highWatermark, lastStableOffset, logStartOffset, Optional.empty(), abortedTransactions, records);
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o)
- return true;
- if (o == null || getClass() != o.getClass())
- return false;
-
- PartitionData that = (PartitionData) o;
-
- return this.partitionResponse.equals(that.partitionResponse);
- }
-
- @Override
- public int hashCode() {
- 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() +
- ", divergingEpoch =" + divergingEpoch() +
- ", recordsSizeInBytes=" + records().sizeInBytes() + ")";
- }
-
- public Errors error() {
- return error;
- }
-
- public long highWatermark() {
- return partitionResponse.highWatermark();
- }
-
- public long lastStableOffset() {
- return partitionResponse.lastStableOffset();
- }
-
- public long logStartOffset() {
- return partitionResponse.logStartOffset();
- }
-
- public Optional<Integer> preferredReadReplica() {
- return preferredReplica;
- }
-
- public List<AbortedTransaction> abortedTransactions() {
- return abortedTransactions;
- }
-
- public Optional<FetchResponseData.EpochEndOffset> divergingEpoch() {
- FetchResponseData.EpochEndOffset epochEndOffset = partitionResponse.divergingEpoch();
- if (epochEndOffset.epoch() < 0) {
- return Optional.empty();
- } else {
- return Optional.of(epochEndOffset);
- }
- }
-
- @SuppressWarnings("unchecked")
- public T records() {
- return (T) partitionResponse.recordSet();
- }
- }
-
- /**
- * From version 3 or later, the entries in `responseData` should be in the same order as the entries in
- * `FetchRequest.fetchData`.
- *
- * @param error The top-level error code.
- * @param responseData The fetched data grouped by partition.
- * @param throttleTimeMs The time in milliseconds that the response was throttled
- * @param sessionId The fetch session id.
- */
public FetchResponse(Errors error,
Review comment:
I think it would be better to make this a static factory method and keep the constructor for the case where we receive `FetchResponseData`.
##########
File path: core/src/test/scala/unit/kafka/server/LogOffsetTest.scala
##########
@@ -127,7 +124,7 @@ class LogOffsetTest extends BaseRequestTest {
Map(topicPartition -> new FetchRequest.PartitionData(consumerOffsets.head, FetchRequest.INVALID_LOG_START_OFFSET,
300 * 1024, Optional.empty())).asJava).build()
val fetchResponse = sendFetchRequest(fetchRequest)
- assertFalse(fetchResponse.responseData.get(topicPartition).records.batches.iterator.hasNext)
+ assertFalse(fetchResponse.responseData.get(topicPartition).records.asInstanceOf[Records].batches.iterator.hasNext)
Review comment:
Similar to elsewhere, it would be useful to have a utility method to avoid unsafe operations all over the code.
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -57,238 +55,51 @@
* the fetch offset after the index lookup
* - {@link Errors#UNKNOWN_SERVER_ERROR} For any unexpected errors
*/
-public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
-
- public static final long INVALID_HIGHWATERMARK = -1L;
+public class FetchResponse extends AbstractResponse {
+ public static final long INVALID_HIGH_WATERMARK = -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 FetchResponseData data;
- private final LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap;
+ private final LinkedHashMap<TopicPartition, FetchResponseData.FetchablePartitionResponse> responseData;
@Override
public FetchResponseData data() {
return data;
}
- public static final class AbortedTransaction {
- public final long producerId;
- public final long firstOffset;
-
- public AbortedTransaction(long producerId, long firstOffset) {
- this.producerId = producerId;
- this.firstOffset = firstOffset;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o)
- return true;
- if (o == null || getClass() != o.getClass())
- return false;
-
- AbortedTransaction that = (AbortedTransaction) o;
-
- return producerId == that.producerId && firstOffset == that.firstOffset;
- }
-
- @Override
- public int hashCode() {
- int result = Long.hashCode(producerId);
- result = 31 * result + Long.hashCode(firstOffset);
- return result;
- }
-
- @Override
- 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<T extends BaseRecords> {
- private final FetchResponseData.FetchablePartitionResponse partitionResponse;
-
- // Derived fields
- private final Optional<Integer> preferredReplica;
- private final List<AbortedTransaction> 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.preferredReadReplica())
- .filter(replicaId -> replicaId != INVALID_PREFERRED_REPLICA_ID);
-
- if (partitionResponse.abortedTransactions() == null) {
- this.abortedTransactions = null;
- } else {
- this.abortedTransactions = partitionResponse.abortedTransactions().stream()
- .map(AbortedTransaction::fromMessage)
- .collect(Collectors.toList());
- }
-
- this.error = Errors.forCode(partitionResponse.errorCode());
- }
-
- public PartitionData(Errors error,
- long highWatermark,
- long lastStableOffset,
- long logStartOffset,
- Optional<Integer> preferredReadReplica,
- List<AbortedTransaction> abortedTransactions,
- Optional<FetchResponseData.EpochEndOffset> divergingEpoch,
- T records) {
- this.preferredReplica = preferredReadReplica;
- this.abortedTransactions = abortedTransactions;
- this.error = error;
-
- FetchResponseData.FetchablePartitionResponse partitionResponse =
- new FetchResponseData.FetchablePartitionResponse();
- partitionResponse.setErrorCode(error.code())
- .setHighWatermark(highWatermark)
- .setLastStableOffset(lastStableOffset)
- .setLogStartOffset(logStartOffset);
- if (abortedTransactions != null) {
- partitionResponse.setAbortedTransactions(abortedTransactions.stream().map(
- aborted -> new FetchResponseData.AbortedTransaction()
- .setProducerId(aborted.producerId)
- .setFirstOffset(aborted.firstOffset))
- .collect(Collectors.toList()));
- } else {
- partitionResponse.setAbortedTransactions(null);
- }
- partitionResponse.setPreferredReadReplica(preferredReadReplica.orElse(INVALID_PREFERRED_REPLICA_ID));
- partitionResponse.setRecordSet(records);
- divergingEpoch.ifPresent(partitionResponse::setDivergingEpoch);
-
- this.partitionResponse = partitionResponse;
- }
-
- public PartitionData(Errors error,
- long highWatermark,
- long lastStableOffset,
- long logStartOffset,
- Optional<Integer> preferredReadReplica,
- List<AbortedTransaction> abortedTransactions,
- T records) {
- this(error, highWatermark, lastStableOffset, logStartOffset, preferredReadReplica,
- abortedTransactions, Optional.empty(), records);
- }
-
- public PartitionData(Errors error,
- long highWatermark,
- long lastStableOffset,
- long logStartOffset,
- List<AbortedTransaction> abortedTransactions,
- T records) {
- this(error, highWatermark, lastStableOffset, logStartOffset, Optional.empty(), abortedTransactions, records);
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o)
- return true;
- if (o == null || getClass() != o.getClass())
- return false;
-
- PartitionData that = (PartitionData) o;
-
- return this.partitionResponse.equals(that.partitionResponse);
- }
-
- @Override
- public int hashCode() {
- 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() +
- ", divergingEpoch =" + divergingEpoch() +
- ", recordsSizeInBytes=" + records().sizeInBytes() + ")";
- }
-
- public Errors error() {
- return error;
- }
-
- public long highWatermark() {
- return partitionResponse.highWatermark();
- }
-
- public long lastStableOffset() {
- return partitionResponse.lastStableOffset();
- }
-
- public long logStartOffset() {
- return partitionResponse.logStartOffset();
- }
-
- public Optional<Integer> preferredReadReplica() {
- return preferredReplica;
- }
-
- public List<AbortedTransaction> abortedTransactions() {
- return abortedTransactions;
- }
-
- public Optional<FetchResponseData.EpochEndOffset> divergingEpoch() {
- FetchResponseData.EpochEndOffset epochEndOffset = partitionResponse.divergingEpoch();
- if (epochEndOffset.epoch() < 0) {
- return Optional.empty();
- } else {
- return Optional.of(epochEndOffset);
- }
- }
-
- @SuppressWarnings("unchecked")
- public T records() {
- return (T) partitionResponse.recordSet();
- }
- }
-
- /**
- * From version 3 or later, the entries in `responseData` should be in the same order as the entries in
- * `FetchRequest.fetchData`.
- *
- * @param error The top-level error code.
- * @param responseData The fetched data grouped by partition.
- * @param throttleTimeMs The time in milliseconds that the response was throttled
- * @param sessionId The fetch session id.
- */
public FetchResponse(Errors error,
- LinkedHashMap<TopicPartition, PartitionData<T>> responseData,
int throttleTimeMs,
- int sessionId) {
- super(ApiKeys.FETCH);
- this.data = toMessage(throttleTimeMs, error, responseData.entrySet().iterator(), sessionId);
- this.responseDataMap = responseData;
+ int sessionId,
+ LinkedHashMap<TopicPartition, FetchResponseData.FetchablePartitionResponse> responseData) {
+ this(new FetchResponseData()
+ .setSessionId(sessionId)
+ .setErrorCode(error.code())
+ .setThrottleTimeMs(throttleTimeMs)
+ .setResponses(responseData.entrySet().stream().map(entry -> new FetchResponseData.FetchableTopicResponse()
+ .setTopic(entry.getKey().topic())
+ .setPartitionResponses(Collections.singletonList(entry.getValue().setPartition(entry.getKey().partition()))))
+ .collect(Collectors.toList())));
}
public FetchResponse(FetchResponseData fetchResponseData) {
super(ApiKeys.FETCH);
this.data = fetchResponseData;
- this.responseDataMap = toResponseDataMap(fetchResponseData);
+ this.responseData = new LinkedHashMap<>();
Review comment:
This isn't needed when we return a fetch from the broker, right? If this is true, can we remove it from the fetch response and built it on the client when needed?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r546203915
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -777,12 +775,12 @@ class KafkaApis(val requestChannel: RequestChannel,
}
def maybeConvertFetchedData(tp: TopicPartition,
Review comment:
@ijuma This is the only case that we create ```LazyDownConversionRecords``` in production. Through this PR, this case can get rid of generic ```FetchResponse.PartitionData```. Hence, we can remove generic from ```FetchResponse.PartitionData``` after this PR goes in trunk.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r587191008
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -365,17 +126,105 @@ public int sessionId() {
* @param partIterator The partition iterator.
* @return The response size in bytes.
*/
- public static <T extends BaseRecords> int sizeOf(short version,
- Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ public static int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, FetchResponseData.PartitionData>> partIterator) {
// Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
// use arbitrary values here without affecting the result.
- FetchResponseData data = toMessage(0, Errors.NONE, partIterator, INVALID_SESSION_ID);
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
Review comment:
Is this an additional copy compared to previous behavior?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r585235586
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -365,17 +126,92 @@ public int sessionId() {
* @param partIterator The partition iterator.
* @return The response size in bytes.
*/
- public static <T extends BaseRecords> int sizeOf(short version,
- Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ public static int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, FetchResponseData.PartitionData>> partIterator) {
// Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
// use arbitrary values here without affecting the result.
- FetchResponseData data = toMessage(0, Errors.NONE, partIterator, INVALID_SESSION_ID);
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
ObjectSerializationCache cache = new ObjectSerializationCache();
- return 4 + data.size(cache, version);
+ return 4 + FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, data).data.size(cache, version);
}
@Override
public boolean shouldClientThrottle(short version) {
return version >= 8;
}
-}
+
+ public static Optional<FetchResponseData.EpochEndOffset> divergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() < 0 ? Optional.empty()
+ : Optional.of(partitionResponse.divergingEpoch());
+ }
+
+ public static boolean isDivergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() >= 0;
+ }
+
+ public static Optional<Integer> preferredReadReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() == INVALID_PREFERRED_REPLICA_ID ? Optional.empty()
+ : Optional.of(partitionResponse.preferredReadReplica());
+ }
+
+ public static boolean isPreferredReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() != INVALID_PREFERRED_REPLICA_ID;
+ }
+
+ public static FetchResponseData.PartitionData partitionResponse(int partition, Errors error) {
+ return new FetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK);
+ }
+
+ /**
+ * cast the BaseRecords of PartitionData to Records. KRPC converts the byte array to MemoryRecords so this method
+ * never fail if the data is from KRPC.
+ *
+ * @param partition partition data
+ * @return Records or empty record if the records in PartitionData is null.
+ */
+ public static Records records(FetchResponseData.PartitionData partition) {
Review comment:
good point. will copy that
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r585229860
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -365,17 +126,92 @@ public int sessionId() {
* @param partIterator The partition iterator.
* @return The response size in bytes.
*/
- public static <T extends BaseRecords> int sizeOf(short version,
- Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ public static int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, FetchResponseData.PartitionData>> partIterator) {
// Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
// use arbitrary values here without affecting the result.
- FetchResponseData data = toMessage(0, Errors.NONE, partIterator, INVALID_SESSION_ID);
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
ObjectSerializationCache cache = new ObjectSerializationCache();
- return 4 + data.size(cache, version);
+ return 4 + FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, data).data.size(cache, version);
}
@Override
public boolean shouldClientThrottle(short version) {
return version >= 8;
}
-}
+
+ public static Optional<FetchResponseData.EpochEndOffset> divergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() < 0 ? Optional.empty()
+ : Optional.of(partitionResponse.divergingEpoch());
+ }
+
+ public static boolean isDivergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() >= 0;
+ }
+
+ public static Optional<Integer> preferredReadReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() == INVALID_PREFERRED_REPLICA_ID ? Optional.empty()
+ : Optional.of(partitionResponse.preferredReadReplica());
+ }
+
+ public static boolean isPreferredReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() != INVALID_PREFERRED_REPLICA_ID;
+ }
+
+ public static FetchResponseData.PartitionData partitionResponse(int partition, Errors error) {
+ return new FetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK);
+ }
+
+ /**
+ * cast the BaseRecords of PartitionData to Records. KRPC converts the byte array to MemoryRecords so this method
+ * never fail if the data is from KRPC.
+ *
+ * @param partition partition data
+ * @return Records or empty record if the records in PartitionData is null.
+ */
+ public static Records records(FetchResponseData.PartitionData partition) {
+ return partition.records() == null ? MemoryRecords.EMPTY : (Records) partition.records();
Review comment:
will copy that
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r551441782
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -59,136 +60,140 @@
*/
public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
- public static final long INVALID_HIGHWATERMARK = -1L;
+ public static FetchResponseData.FetchablePartitionResponse partitionResponse(Errors error) {
+ return new FetchResponseData.FetchablePartitionResponse()
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK)
+ .setLastStableOffset(FetchResponse.INVALID_LAST_STABLE_OFFSET)
+ .setLogStartOffset(FetchResponse.INVALID_LOG_START_OFFSET)
+ .setAbortedTransactions(null)
+ .setRecordSet(MemoryRecords.EMPTY)
+ .setPreferredReadReplica(FetchResponse.INVALID_PREFERRED_REPLICA_ID);
+ }
+
+ public static final long INVALID_HIGH_WATERMARK = -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 FetchResponseData data;
- private final LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap;
+ private final LinkedHashMap<TopicPartition, FetchResponseData.FetchablePartitionResponse> partitionData;
+ // lazily instantiate this field
+ private LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap = null;
Review comment:
```partitionData``` can be removed but ```responseDataMap``` is still a response type and it is used by production code. I did not refactor that to avoid big patch.
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -59,136 +60,140 @@
*/
public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
- public static final long INVALID_HIGHWATERMARK = -1L;
+ public static FetchResponseData.FetchablePartitionResponse partitionResponse(Errors error) {
+ return new FetchResponseData.FetchablePartitionResponse()
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK)
+ .setLastStableOffset(FetchResponse.INVALID_LAST_STABLE_OFFSET)
+ .setLogStartOffset(FetchResponse.INVALID_LOG_START_OFFSET)
+ .setAbortedTransactions(null)
+ .setRecordSet(MemoryRecords.EMPTY)
+ .setPreferredReadReplica(FetchResponse.INVALID_PREFERRED_REPLICA_ID);
+ }
+
+ public static final long INVALID_HIGH_WATERMARK = -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 FetchResponseData data;
- private final LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap;
+ private final LinkedHashMap<TopicPartition, FetchResponseData.FetchablePartitionResponse> partitionData;
+ // lazily instantiate this field
+ private LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap = null;
@Override
public FetchResponseData data() {
return data;
}
- public static final class AbortedTransaction {
- public final long producerId;
- public final long firstOffset;
+ /**
+ * From version 3 or later, the entries in `responseData` should be in the same order as the entries in
+ * `FetchRequest.fetchData`.
+ *
+ * @param error The top-level error code.
+ * @param responseData The fetched data grouped by partition.
+ * @param throttleTimeMs The time in milliseconds that the response was throttled
+ * @param sessionId The fetch session id.
+ */
+ public FetchResponse(Errors error,
+ LinkedHashMap<TopicPartition, PartitionData<T>> responseData,
+ int throttleTimeMs,
+ int sessionId) {
+ this(error, throttleTimeMs, sessionId, responseData.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey,
+ entry -> entry.getValue().partitionResponse, (o1, o2) -> {
+ throw new RuntimeException("this is impossible");
+ }, LinkedHashMap::new)));
+ }
- public AbortedTransaction(long producerId, long firstOffset) {
- this.producerId = producerId;
- this.firstOffset = firstOffset;
- }
+ public FetchResponse(Errors error,
+ int throttleTimeMs,
+ int sessionId,
+ LinkedHashMap<TopicPartition, FetchResponseData.FetchablePartitionResponse> responseData) {
+ super(ApiKeys.FETCH);
+ this.data = new FetchResponseData()
+ .setSessionId(sessionId)
+ .setErrorCode(error.code())
+ .setThrottleTimeMs(throttleTimeMs);
+ responseData.forEach((tp, tpData) -> data.responses().add(new FetchResponseData.FetchableTopicResponse()
+ .setTopic(tp.topic())
+ .setPartitionResponses(Collections.singletonList(tpData.setPartition(tp.partition())))));
+ this.partitionData = responseData;
+ }
- @Override
- public boolean equals(Object o) {
- if (this == o)
- return true;
- if (o == null || getClass() != o.getClass())
- return false;
+ public FetchResponse(FetchResponseData fetchResponseData) {
+ super(ApiKeys.FETCH);
+ this.data = fetchResponseData;
+ this.partitionData = new LinkedHashMap<>();
+ fetchResponseData.responses().forEach(topicResponse ->
+ topicResponse.partitionResponses().forEach(partitionResponse ->
+ partitionData.put(new TopicPartition(topicResponse.topic(), partitionResponse.partition()), partitionResponse))
+ );
+ }
- AbortedTransaction that = (AbortedTransaction) o;
+ public Errors error() {
+ return Errors.forCode(data.errorCode());
+ }
- return producerId == that.producerId && firstOffset == that.firstOffset;
+ public LinkedHashMap<TopicPartition, PartitionData<T>> responseData() {
+ if (responseDataMap == null) {
+ responseDataMap = new LinkedHashMap<>(partitionData.size());
+ partitionData.forEach((tp, d) -> responseDataMap.put(tp, new PartitionData<>(d)));
}
+ return responseDataMap;
+ }
- @Override
- public int hashCode() {
- int result = Long.hashCode(producerId);
- result = 31 * result + Long.hashCode(firstOffset);
- return result;
- }
+ @Override
+ public int throttleTimeMs() {
+ return data.throttleTimeMs();
+ }
- @Override
- public String toString() {
- return "(producerId=" + producerId + ", firstOffset=" + firstOffset + ")";
- }
+ public int sessionId() {
+ return data.sessionId();
+ }
- static AbortedTransaction fromMessage(FetchResponseData.AbortedTransaction abortedTransaction) {
- return new AbortedTransaction(abortedTransaction.producerId(), abortedTransaction.firstOffset());
- }
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ Map<Errors, Integer> errorCounts = new HashMap<>();
+ updateErrorCounts(errorCounts, error());
+ partitionData.values().forEach(response ->
+ updateErrorCounts(errorCounts, Errors.forCode(response.errorCode()))
+ );
+ return errorCounts;
}
- public static final class PartitionData<T extends BaseRecords> {
- private final FetchResponseData.FetchablePartitionResponse partitionResponse;
+ public static FetchResponse<MemoryRecords> parse(ByteBuffer buffer, short version) {
+ return new FetchResponse<>(new FetchResponseData(new ByteBufferAccessor(buffer), version));
+ }
- // Derived fields
- private final Optional<Integer> preferredReplica;
- private final List<AbortedTransaction> abortedTransactions;
- private final Errors error;
+ /**
+ * Convenience method to find the size of a response.
+ *
+ * @param version The version of the response to use.
+ * @param partIterator The partition iterator.
+ * @return The response size in bytes.
+ */
+ public static <T extends Records> int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ // Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
+ // use arbitrary values here without affecting the result.
+ LinkedHashMap<TopicPartition, PartitionData<T>> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
+ ObjectSerializationCache cache = new ObjectSerializationCache();
+ return 4 + new FetchResponse<>(Errors.NONE, data, 0, INVALID_SESSION_ID).data.size(cache, version);
+ }
- 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.preferredReadReplica())
- .filter(replicaId -> replicaId != INVALID_PREFERRED_REPLICA_ID);
-
- if (partitionResponse.abortedTransactions() == null) {
- this.abortedTransactions = null;
- } else {
- this.abortedTransactions = partitionResponse.abortedTransactions().stream()
- .map(AbortedTransaction::fromMessage)
- .collect(Collectors.toList());
- }
-
- this.error = Errors.forCode(partitionResponse.errorCode());
- }
+ @Override
+ public boolean shouldClientThrottle(short version) {
+ return version >= 8;
+ }
- public PartitionData(Errors error,
- long highWatermark,
- long lastStableOffset,
- long logStartOffset,
- Optional<Integer> preferredReadReplica,
- List<AbortedTransaction> abortedTransactions,
- Optional<FetchResponseData.EpochEndOffset> divergingEpoch,
- T records) {
- this.preferredReplica = preferredReadReplica;
- this.abortedTransactions = abortedTransactions;
- this.error = error;
-
- FetchResponseData.FetchablePartitionResponse partitionResponse =
- new FetchResponseData.FetchablePartitionResponse();
- partitionResponse.setErrorCode(error.code())
- .setHighWatermark(highWatermark)
- .setLastStableOffset(lastStableOffset)
- .setLogStartOffset(logStartOffset);
- if (abortedTransactions != null) {
- partitionResponse.setAbortedTransactions(abortedTransactions.stream().map(
- aborted -> new FetchResponseData.AbortedTransaction()
- .setProducerId(aborted.producerId)
- .setFirstOffset(aborted.firstOffset))
- .collect(Collectors.toList()));
- } else {
- partitionResponse.setAbortedTransactions(null);
- }
- partitionResponse.setPreferredReadReplica(preferredReadReplica.orElse(INVALID_PREFERRED_REPLICA_ID));
- partitionResponse.setRecordSet(records);
- divergingEpoch.ifPresent(partitionResponse::setDivergingEpoch);
- this.partitionResponse = partitionResponse;
- }
+ public static final class PartitionData<T extends BaseRecords> {
Review comment:
> What does it add over FetchResponseData.FetchablePartitionRespons?
It offers methods to transfer "non-defined value" to ```Optional``` type.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584225973
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -304,58 +115,12 @@ public int sessionId() {
public Map<Errors, Integer> errorCounts() {
Map<Errors, Integer> errorCounts = new HashMap<>();
updateErrorCounts(errorCounts, error());
- responseDataMap.values().forEach(response ->
- updateErrorCounts(errorCounts, response.error())
- );
+ dataByTopicPartition.values().forEach(response -> updateErrorCounts(errorCounts, Errors.forCode(response.errorCode())));
return errorCounts;
}
- public static FetchResponse<MemoryRecords> parse(ByteBuffer buffer, short version) {
- return new FetchResponse<>(new FetchResponseData(new ByteBufferAccessor(buffer), version));
- }
-
- @SuppressWarnings("unchecked")
- private static <T extends BaseRecords> LinkedHashMap<TopicPartition, PartitionData<T>> toResponseDataMap(
- FetchResponseData message) {
- LinkedHashMap<TopicPartition, PartitionData<T>> responseMap = new LinkedHashMap<>();
- message.responses().forEach(topicResponse -> {
- topicResponse.partitionResponses().forEach(partitionResponse -> {
- TopicPartition tp = new TopicPartition(topicResponse.topic(), partitionResponse.partition());
- PartitionData<T> partitionData = new PartitionData<>(partitionResponse);
- responseMap.put(tp, partitionData);
- });
- });
- return responseMap;
- }
-
- private static <T extends BaseRecords> FetchResponseData toMessage(int throttleTimeMs, Errors error,
Review comment:
Oh, I planed to remove all usages of this method from production (i.e `KafkaApis` should generate batched response directly). However, it can produce a big patch so I will keep this method in next commit.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584141887
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -57,238 +56,39 @@
* the fetch offset after the index lookup
* - {@link Errors#UNKNOWN_SERVER_ERROR} For any unexpected errors
*/
-public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
-
- public static final long INVALID_HIGHWATERMARK = -1L;
+public class FetchResponse extends AbstractResponse {
+ public static final long INVALID_HIGH_WATERMARK = -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 FetchResponseData data;
- private final LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap;
+ // we build responseData when needed.
+ private LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData = null;
Review comment:
This is not thread-safe and requests are typically thread-safe. What's the thinking here?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584221983
##########
File path: core/src/main/scala/kafka/server/AbstractFetcherThread.scala
##########
@@ -734,7 +730,7 @@ abstract class AbstractFetcherThread(name: String,
Option(partitionStates.stateValue(topicPartition))
}
- protected def toMemoryRecords(records: Records): MemoryRecords = {
+ protected def toMemoryRecords(records: BaseRecords): MemoryRecords = {
Review comment:
This change can be reverted as we have helper method `FetchResponse#records` now.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-789031053
@ijuma thanks for your comments. I have addressed them in latest commit (https://github.com/apache/kafka/pull/9758/commits/55b35ab5764d8129e23f386229eb350603b7b0aa). Please take a look.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-788567775
@ijuma Thanks for all your great comments. I have updated this PR. Please take a look.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584222449
##########
File path: jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetchsession/FetchSessionBenchmark.java
##########
@@ -70,24 +70,25 @@ public void setUp() {
handler = new FetchSessionHandler(LOG_CONTEXT, 1);
FetchSessionHandler.Builder builder = handler.newBuilder();
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> respMap = new LinkedHashMap<>();
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> respMap = new LinkedHashMap<>();
for (int i = 0; i < partitionCount; i++) {
TopicPartition tp = new TopicPartition("foo", i);
FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData(0, 0, 200,
Optional.empty());
fetches.put(tp, partitionData);
builder.add(tp, partitionData);
- respMap.put(tp, new FetchResponse.PartitionData<>(
- Errors.NONE,
- 0L,
- 0L,
- 0,
- null,
- null));
+ respMap.put(tp, new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition())
+ .setErrorCode(Errors.NONE.code())
+ .setHighWatermark(0)
+ .setLastStableOffset(0)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(null)
+ .setRecords(null));
Review comment:
updated
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584224750
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -761,76 +754,80 @@ class KafkaApis(val requestChannel: RequestChannel,
// For fetch requests from clients, check if down-conversion is disabled for the particular partition
if (!fetchRequest.isFromFollower && !logConfig.forall(_.messageDownConversionEnable)) {
trace(s"Conversion to message format ${downConvertMagic.get} is disabled for partition $tp. Sending unsupported version response to $clientId.")
- errorResponse(Errors.UNSUPPORTED_VERSION)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_VERSION)
} else {
try {
trace(s"Down converting records from partition $tp to message format version $magic for fetch request from $clientId")
// Because down-conversion is extremely memory intensive, we want to try and delay the down-conversion as much
// 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.
- val error = maybeDownConvertStorageError(partitionData.error)
- 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))
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(new LazyDownConversionRecords(tp, unconvertedRecords, magic, fetchContext.getFetchOffset(tp).get, time))
+ .setPreferredReadReplica(partitionData.preferredReadReplica())
} catch {
case e: UnsupportedCompressionTypeException =>
trace("Received unsupported compression type error during down-conversion", e)
- errorResponse(Errors.UNSUPPORTED_COMPRESSION_TYPE)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_COMPRESSION_TYPE)
}
}
case None =>
- val error = maybeDownConvertStorageError(partitionData.error)
- new FetchResponse.PartitionData[BaseRecords](error,
- partitionData.highWatermark,
- partitionData.lastStableOffset,
- partitionData.logStartOffset,
- partitionData.preferredReadReplica,
- partitionData.abortedTransactions,
- partitionData.divergingEpoch,
- unconvertedRecords)
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(unconvertedRecords)
+ .setPreferredReadReplica(partitionData.preferredReadReplica)
+ .setDivergingEpoch(partitionData.divergingEpoch)
Review comment:
same to https://github.com/apache/kafka/pull/9758#discussion_r584224693
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584165033
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -57,238 +56,39 @@
* the fetch offset after the index lookup
* - {@link Errors#UNKNOWN_SERVER_ERROR} For any unexpected errors
*/
-public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
-
- public static final long INVALID_HIGHWATERMARK = -1L;
+public class FetchResponse extends AbstractResponse {
+ public static final long INVALID_HIGH_WATERMARK = -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 FetchResponseData data;
- private final LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap;
+ // we build responseData when needed.
+ private LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData = null;
Review comment:
jira: https://issues.apache.org/jira/browse/KAFKA-12385
> you can use a volatile field and synchronize on the assignment if still null.
updated
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jolshan commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
jolshan commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-794320479
So should we go forward with https://github.com/apache/kafka/pull/10269 first still?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584812552
##########
File path: clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
##########
@@ -791,15 +791,18 @@ public void produceRequestGetErrorResponseTest() {
@Test
public void fetchResponseVersionTest() {
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> responseData = new LinkedHashMap<>();
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData = new LinkedHashMap<>();
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(), Collections.emptyList(), records));
-
- FetchResponse<MemoryRecords> v0Response = new FetchResponse<>(Errors.NONE, responseData, 0, INVALID_SESSION_ID);
- FetchResponse<MemoryRecords> v1Response = new FetchResponse<>(Errors.NONE, responseData, 10, INVALID_SESSION_ID);
+ responseData.put(new TopicPartition("test", 0),
+ new FetchResponseData.PartitionData()
+ .setHighWatermark(1000000)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(Collections.emptyList())
Review comment:
Aborted transactions is empty by default.
##########
File path: clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
##########
@@ -1146,38 +1159,47 @@ private FetchRequest createFetchRequest(int version) {
return FetchRequest.Builder.forConsumer(100, 100000, fetchData).setMaxBytes(1000).build((short) version);
}
- private FetchResponse<MemoryRecords> createFetchResponse(Errors error, int sessionId) {
- return new FetchResponse<>(error, new LinkedHashMap<>(), 25, sessionId);
+ private FetchResponse createFetchResponse(Errors error, int sessionId) {
+ return FetchResponse.of(error, 25, sessionId, new LinkedHashMap<>());
}
- private FetchResponse<MemoryRecords> createFetchResponse(int sessionId) {
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> responseData = new LinkedHashMap<>();
+ private FetchResponse createFetchResponse(int sessionId) {
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> 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(), Collections.emptyList(), records));
- List<FetchResponse.AbortedTransaction> 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));
- return new FetchResponse<>(Errors.NONE, responseData, 25, sessionId);
- }
-
- private FetchResponse<MemoryRecords> createFetchResponse(boolean includeAborted) {
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> responseData = new LinkedHashMap<>();
+ responseData.put(new TopicPartition("test", 0), new FetchResponseData.PartitionData()
+ .setHighWatermark(1000000)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(Collections.emptyList())
Review comment:
No need to set aborted transactions.
##########
File path: core/src/main/scala/kafka/server/AbstractFetcherThread.scala
##########
@@ -416,9 +412,8 @@ abstract class AbstractFetcherThread(name: String,
"expected to persist.")
partitionsWithError += topicPartition
- case _ =>
- error(s"Error for partition $topicPartition at offset ${currentFetchState.fetchOffset}",
- partitionData.error.exception)
+ case partitionError: Errors =>
Review comment:
We should not have `: Errors` here as it introduces a type test. What we want is for this to be a catch all.
##########
File path: core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala
##########
@@ -1144,8 +1143,14 @@ class AbstractFetcherThreadTest {
(Errors.NONE, records)
}
- (partition, new FetchData(error, leaderState.highWatermark, leaderState.highWatermark, leaderState.logStartOffset,
- Optional.empty[Integer], List.empty.asJava, divergingEpoch.asJava, records))
+ (partition, new FetchResponseData.PartitionData()
+ .setErrorCode(error.code)
+ .setHighWatermark(leaderState.highWatermark)
+ .setLastStableOffset(leaderState.highWatermark)
+ .setLogStartOffset(leaderState.logStartOffset)
+ .setAbortedTransactions(Collections.emptyList())
Review comment:
Redundant.
##########
File path: clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
##########
@@ -808,22 +811,32 @@ public void fetchResponseVersionTest() {
@Test
public void testFetchResponseV4() {
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> responseData = new LinkedHashMap<>();
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData = new LinkedHashMap<>();
MemoryRecords records = MemoryRecords.readableRecords(ByteBuffer.allocate(10));
- List<FetchResponse.AbortedTransaction> abortedTransactions = asList(
- new FetchResponse.AbortedTransaction(10, 100),
- new FetchResponse.AbortedTransaction(15, 50)
+ List<FetchResponseData.AbortedTransaction> abortedTransactions = asList(
+ new FetchResponseData.AbortedTransaction().setProducerId(10).setFirstOffset(100),
+ new FetchResponseData.AbortedTransaction().setProducerId(15).setFirstOffset(50)
);
- responseData.put(new TopicPartition("bar", 0), new FetchResponse.PartitionData<>(Errors.NONE, 100000,
- FetchResponse.INVALID_LAST_STABLE_OFFSET, FetchResponse.INVALID_LOG_START_OFFSET, Optional.empty(), abortedTransactions, records));
- responseData.put(new TopicPartition("bar", 1), new FetchResponse.PartitionData<>(Errors.NONE, 900000,
- 5, FetchResponse.INVALID_LOG_START_OFFSET, Optional.empty(), null, records));
- responseData.put(new TopicPartition("foo", 0), new FetchResponse.PartitionData<>(Errors.NONE, 70000,
- 6, FetchResponse.INVALID_LOG_START_OFFSET, Optional.empty(), emptyList(), records));
-
- FetchResponse<MemoryRecords> response = new FetchResponse<>(Errors.NONE, responseData, 10, INVALID_SESSION_ID);
- FetchResponse<MemoryRecords> deserialized = FetchResponse.parse(response.serialize((short) 4), (short) 4);
+ responseData.put(new TopicPartition("bar", 0),
+ new FetchResponseData.PartitionData()
+ .setHighWatermark(1000000)
+ .setAbortedTransactions(abortedTransactions)
+ .setRecords(records));
+ responseData.put(new TopicPartition("bar", 1),
+ new FetchResponseData.PartitionData()
Review comment:
Do we need to set the partition id here? There are a few other cases in this file that are similar.
##########
File path: clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
##########
@@ -1146,38 +1159,47 @@ private FetchRequest createFetchRequest(int version) {
return FetchRequest.Builder.forConsumer(100, 100000, fetchData).setMaxBytes(1000).build((short) version);
}
- private FetchResponse<MemoryRecords> createFetchResponse(Errors error, int sessionId) {
- return new FetchResponse<>(error, new LinkedHashMap<>(), 25, sessionId);
+ private FetchResponse createFetchResponse(Errors error, int sessionId) {
+ return FetchResponse.of(error, 25, sessionId, new LinkedHashMap<>());
}
- private FetchResponse<MemoryRecords> createFetchResponse(int sessionId) {
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> responseData = new LinkedHashMap<>();
+ private FetchResponse createFetchResponse(int sessionId) {
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> 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(), Collections.emptyList(), records));
- List<FetchResponse.AbortedTransaction> 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));
- return new FetchResponse<>(Errors.NONE, responseData, 25, sessionId);
- }
-
- private FetchResponse<MemoryRecords> createFetchResponse(boolean includeAborted) {
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> responseData = new LinkedHashMap<>();
+ responseData.put(new TopicPartition("test", 0), new FetchResponseData.PartitionData()
+ .setHighWatermark(1000000)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(Collections.emptyList())
+ .setRecords(records));
+ List<FetchResponseData.AbortedTransaction> abortedTransactions = Collections.singletonList(
+ new FetchResponseData.AbortedTransaction().setProducerId(234L).setFirstOffset(999L));
+ responseData.put(new TopicPartition("test", 1), new FetchResponseData.PartitionData()
+ .setHighWatermark(1000000)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(abortedTransactions));
+ return FetchResponse.of(Errors.NONE, 25, sessionId, responseData);
+ }
+
+ private FetchResponse createFetchResponse(boolean includeAborted) {
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData = new LinkedHashMap<>();
MemoryRecords records = MemoryRecords.withRecords(CompressionType.NONE, new SimpleRecord("blah".getBytes()));
+ responseData.put(new TopicPartition("test", 0), new FetchResponseData.PartitionData()
+ .setHighWatermark(1000000)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(Collections.emptyList())
Review comment:
Aborted transactions is empty by default.
##########
File path: core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala
##########
@@ -1144,8 +1143,14 @@ class AbstractFetcherThreadTest {
(Errors.NONE, records)
}
- (partition, new FetchData(error, leaderState.highWatermark, leaderState.highWatermark, leaderState.logStartOffset,
- Optional.empty[Integer], List.empty.asJava, divergingEpoch.asJava, records))
+ (partition, new FetchResponseData.PartitionData()
+ .setErrorCode(error.code)
+ .setHighWatermark(leaderState.highWatermark)
+ .setLastStableOffset(leaderState.highWatermark)
+ .setLogStartOffset(leaderState.logStartOffset)
+ .setAbortedTransactions(Collections.emptyList())
+ .setRecords(records)
+ .setDivergingEpoch(divergingEpoch.getOrElse(new FetchResponseData.EpochEndOffset)))
Review comment:
The previous code did `asJava`, why did we change it?
##########
File path: jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetcher/ReplicaFetcherThreadBenchmark.java
##########
@@ -174,8 +173,12 @@ public int sizeInBytes() {
return null;
}
};
- initialFetched.put(tp, new FetchResponse.PartitionData<>(Errors.NONE, 0, 0, 0,
- new LinkedList<>(), fetched));
+ initialFetched.put(tp, new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition())
+ .setLastStableOffset(0)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(Collections.emptyList())
Review comment:
Redundant.
##########
File path: core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala
##########
@@ -963,9 +963,11 @@ class ReplicaFetcherThreadTest {
val records = MemoryRecords.withRecords(CompressionType.NONE,
new SimpleRecord(1000, "foo".getBytes(StandardCharsets.UTF_8)))
-
- val partitionData: thread.FetchData = new FetchResponse.PartitionData[Records](
- Errors.NONE, 0, 0, 0, Optional.empty(), Collections.emptyList(), records)
+ val partitionData: thread.FetchData = new FetchResponseData.PartitionData()
+ .setLastStableOffset(0)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(Collections.emptyList())
Review comment:
Redundant.
##########
File path: core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala
##########
@@ -531,10 +529,12 @@ class ReplicaFetcherThreadTest {
assertEquals(1, mockNetwork.fetchCount)
partitions.foreach { tp => assertEquals(Fetching, thread.fetchState(tp).get.state) }
- def partitionData(divergingEpoch: FetchResponseData.EpochEndOffset): FetchResponse.PartitionData[Records] = {
- new FetchResponse.PartitionData[Records](
- Errors.NONE, 0, 0, 0, Optional.empty(), Collections.emptyList(),
- Optional.of(divergingEpoch), MemoryRecords.EMPTY)
+ def partitionData(divergingEpoch: FetchResponseData.EpochEndOffset): FetchResponseData.PartitionData = {
+ new FetchResponseData.PartitionData()
+ .setLastStableOffset(0)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(Collections.emptyList())
Review comment:
Redundant.
##########
File path: core/src/test/scala/unit/kafka/server/FetchSessionTest.scala
##########
@@ -658,12 +732,19 @@ class FetchSessionTest {
// Full fetch context returns all partitions in the response
val context1 = fetchManager.newContext(JFetchMetadata.INITIAL, reqData, EMPTY_PART_LIST, isFollower = false)
assertEquals(classOf[FullFetchContext], context1.getClass)
- val respData = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[Records]]
- respData.put(tp1, new FetchResponse.PartitionData(Errors.NONE,
- 105, 105, 0, Optional.empty(), Collections.emptyList(), Optional.empty(), null))
- val divergingEpoch = Optional.of(new FetchResponseData.EpochEndOffset().setEpoch(3).setEndOffset(90))
- respData.put(tp2, new FetchResponse.PartitionData(Errors.NONE,
- 105, 105, 0, Optional.empty(), Collections.emptyList(), divergingEpoch, null))
+ val respData = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
+ respData.put(tp1, new FetchResponseData.PartitionData()
+ .setHighWatermark(105)
+ .setLastStableOffset(105)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(Collections.emptyList()))
+ val divergingEpoch = new FetchResponseData.EpochEndOffset().setEpoch(3).setEndOffset(90)
+ respData.put(tp2, new FetchResponseData.PartitionData()
+ .setHighWatermark(105)
+ .setLastStableOffset(105)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(Collections.emptyList())
Review comment:
Redundant.
##########
File path: core/src/test/scala/unit/kafka/server/FetchSessionTest.scala
##########
@@ -679,17 +760,24 @@ class FetchSessionTest {
assertEquals(Collections.singleton(tp2), resp2.responseData.keySet)
// All partitions with divergent epoch should be returned.
- respData.put(tp1, new FetchResponse.PartitionData(Errors.NONE,
- 105, 105, 0, Optional.empty(), Collections.emptyList(), divergingEpoch, null))
+ respData.put(tp1, new FetchResponseData.PartitionData()
+ .setHighWatermark(105)
+ .setLastStableOffset(105)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(Collections.emptyList())
Review comment:
Redundant.
##########
File path: core/src/test/scala/unit/kafka/server/FetchSessionTest.scala
##########
@@ -679,17 +760,24 @@ class FetchSessionTest {
assertEquals(Collections.singleton(tp2), resp2.responseData.keySet)
// All partitions with divergent epoch should be returned.
- respData.put(tp1, new FetchResponse.PartitionData(Errors.NONE,
- 105, 105, 0, Optional.empty(), Collections.emptyList(), divergingEpoch, null))
+ respData.put(tp1, new FetchResponseData.PartitionData()
+ .setHighWatermark(105)
+ .setLastStableOffset(105)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(Collections.emptyList())
+ .setDivergingEpoch(divergingEpoch))
val resp3 = context2.updateAndGenerateResponseData(respData)
assertEquals(Errors.NONE, resp3.error)
assertEquals(resp1.sessionId, resp3.sessionId)
assertEquals(Utils.mkSet(tp1, tp2), resp3.responseData.keySet)
// Partitions that meet other conditions should be returned regardless of whether
// divergingEpoch is set or not.
- respData.put(tp1, new FetchResponse.PartitionData(Errors.NONE,
- 110, 110, 0, Optional.empty(), Collections.emptyList(), Optional.empty(), null))
+ respData.put(tp1, new FetchResponseData.PartitionData()
+ .setHighWatermark(110)
+ .setLastStableOffset(110)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(Collections.emptyList()))
Review comment:
Redundant.
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -365,17 +126,92 @@ public int sessionId() {
* @param partIterator The partition iterator.
* @return The response size in bytes.
*/
- public static <T extends BaseRecords> int sizeOf(short version,
- Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ public static int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, FetchResponseData.PartitionData>> partIterator) {
// Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
// use arbitrary values here without affecting the result.
- FetchResponseData data = toMessage(0, Errors.NONE, partIterator, INVALID_SESSION_ID);
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
ObjectSerializationCache cache = new ObjectSerializationCache();
- return 4 + data.size(cache, version);
+ return 4 + FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, data).data.size(cache, version);
}
@Override
public boolean shouldClientThrottle(short version) {
return version >= 8;
}
-}
+
+ public static Optional<FetchResponseData.EpochEndOffset> divergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() < 0 ? Optional.empty()
+ : Optional.of(partitionResponse.divergingEpoch());
+ }
+
+ public static boolean isDivergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() >= 0;
+ }
+
+ public static Optional<Integer> preferredReadReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() == INVALID_PREFERRED_REPLICA_ID ? Optional.empty()
+ : Optional.of(partitionResponse.preferredReadReplica());
+ }
+
+ public static boolean isPreferredReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() != INVALID_PREFERRED_REPLICA_ID;
+ }
+
+ public static FetchResponseData.PartitionData partitionResponse(int partition, Errors error) {
+ return new FetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK);
+ }
+
+ /**
+ * cast the BaseRecords of PartitionData to Records. KRPC converts the byte array to MemoryRecords so this method
+ * never fail if the data is from KRPC.
Review comment:
Suggestion:
```text
Returns `partition.records` as `Records` (instead of `BaseRecords`). If `records` is `null`, returns `MemoryRecords.EMPTY`.
If this response was deserialized after a fetch, this method should never fail. An example where this would fail is a down-converted response (e.g. LazyDownConversionRecords) on the broker (before it's serialized and sent on the wire).
```
```
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -365,17 +126,92 @@ public int sessionId() {
* @param partIterator The partition iterator.
* @return The response size in bytes.
*/
- public static <T extends BaseRecords> int sizeOf(short version,
- Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ public static int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, FetchResponseData.PartitionData>> partIterator) {
// Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
// use arbitrary values here without affecting the result.
- FetchResponseData data = toMessage(0, Errors.NONE, partIterator, INVALID_SESSION_ID);
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
ObjectSerializationCache cache = new ObjectSerializationCache();
- return 4 + data.size(cache, version);
+ return 4 + FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, data).data.size(cache, version);
}
@Override
public boolean shouldClientThrottle(short version) {
return version >= 8;
}
-}
+
+ public static Optional<FetchResponseData.EpochEndOffset> divergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() < 0 ? Optional.empty()
+ : Optional.of(partitionResponse.divergingEpoch());
+ }
+
+ public static boolean isDivergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() >= 0;
+ }
+
+ public static Optional<Integer> preferredReadReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() == INVALID_PREFERRED_REPLICA_ID ? Optional.empty()
+ : Optional.of(partitionResponse.preferredReadReplica());
+ }
+
+ public static boolean isPreferredReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() != INVALID_PREFERRED_REPLICA_ID;
+ }
+
+ public static FetchResponseData.PartitionData partitionResponse(int partition, Errors error) {
+ return new FetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK);
+ }
+
+ /**
+ * cast the BaseRecords of PartitionData to Records. KRPC converts the byte array to MemoryRecords so this method
+ * never fail if the data is from KRPC.
+ *
+ * @param partition partition data
+ * @return Records or empty record if the records in PartitionData is null.
+ */
+ public static Records records(FetchResponseData.PartitionData partition) {
+ return partition.records() == null ? MemoryRecords.EMPTY : (Records) partition.records();
Review comment:
Instead of casting blindly, can we include a reasonable error message if the cast fails?
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -761,79 +754,84 @@ class KafkaApis(val requestChannel: RequestChannel,
// For fetch requests from clients, check if down-conversion is disabled for the particular partition
if (!fetchRequest.isFromFollower && !logConfig.forall(_.messageDownConversionEnable)) {
trace(s"Conversion to message format ${downConvertMagic.get} is disabled for partition $tp. Sending unsupported version response to $clientId.")
- errorResponse(Errors.UNSUPPORTED_VERSION)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_VERSION)
} else {
try {
trace(s"Down converting records from partition $tp to message format version $magic for fetch request from $clientId")
// Because down-conversion is extremely memory intensive, we want to try and delay the down-conversion as much
// 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.
- val error = maybeDownConvertStorageError(partitionData.error)
- 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))
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(new LazyDownConversionRecords(tp, unconvertedRecords, magic, fetchContext.getFetchOffset(tp).get, time))
+ .setPreferredReadReplica(partitionData.preferredReadReplica())
} catch {
case e: UnsupportedCompressionTypeException =>
trace("Received unsupported compression type error during down-conversion", e)
- errorResponse(Errors.UNSUPPORTED_COMPRESSION_TYPE)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_COMPRESSION_TYPE)
}
}
case None =>
- val error = maybeDownConvertStorageError(partitionData.error)
- new FetchResponse.PartitionData[BaseRecords](error,
- partitionData.highWatermark,
- partitionData.lastStableOffset,
- partitionData.logStartOffset,
- partitionData.preferredReadReplica,
- partitionData.abortedTransactions,
- partitionData.divergingEpoch,
- unconvertedRecords)
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(unconvertedRecords)
+ .setPreferredReadReplica(partitionData.preferredReadReplica)
+ .setDivergingEpoch(partitionData.divergingEpoch)
}
}
}
// the callback for process a fetch response, invoked before throttling
def processResponseCallback(responsePartitionData: Seq[(TopicPartition, FetchPartitionData)]): Unit = {
- val partitions = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[Records]]
+ val partitions = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
val reassigningPartitions = mutable.Set[TopicPartition]()
responsePartitionData.foreach { case (tp, data) =>
val abortedTransactions = data.abortedTransactions.map(_.asJava).orNull
val lastStableOffset = data.lastStableOffset.getOrElse(FetchResponse.INVALID_LAST_STABLE_OFFSET)
- if (data.isReassignmentFetch)
- reassigningPartitions.add(tp)
- val error = maybeDownConvertStorageError(data.error)
- partitions.put(tp, new FetchResponse.PartitionData(
- error,
- data.highWatermark,
- lastStableOffset,
- data.logStartOffset,
- data.preferredReadReplica.map(int2Integer).asJava,
- abortedTransactions,
- data.divergingEpoch.asJava,
- data.records))
+ if (data.isReassignmentFetch) reassigningPartitions.add(tp)
+ partitions.put(tp, new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(data.error).code)
+ .setHighWatermark(data.highWatermark)
+ .setLastStableOffset(lastStableOffset)
+ .setLogStartOffset(data.logStartOffset)
+ .setAbortedTransactions(abortedTransactions)
+ .setRecords(data.records)
+ .setPreferredReadReplica(data.preferredReadReplica.getOrElse(FetchResponse.INVALID_PREFERRED_REPLICA_ID))
+ .setDivergingEpoch(data.divergingEpoch.getOrElse(new FetchResponseData.EpochEndOffset)))
Review comment:
Seems that we could set the diverging offset only if set and leave the default otherwise.
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -365,17 +126,92 @@ public int sessionId() {
* @param partIterator The partition iterator.
* @return The response size in bytes.
*/
- public static <T extends BaseRecords> int sizeOf(short version,
- Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ public static int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, FetchResponseData.PartitionData>> partIterator) {
// Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
// use arbitrary values here without affecting the result.
- FetchResponseData data = toMessage(0, Errors.NONE, partIterator, INVALID_SESSION_ID);
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
ObjectSerializationCache cache = new ObjectSerializationCache();
- return 4 + data.size(cache, version);
+ return 4 + FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, data).data.size(cache, version);
}
@Override
public boolean shouldClientThrottle(short version) {
return version >= 8;
}
-}
+
+ public static Optional<FetchResponseData.EpochEndOffset> divergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() < 0 ? Optional.empty()
+ : Optional.of(partitionResponse.divergingEpoch());
+ }
+
+ public static boolean isDivergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() >= 0;
+ }
+
+ public static Optional<Integer> preferredReadReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() == INVALID_PREFERRED_REPLICA_ID ? Optional.empty()
+ : Optional.of(partitionResponse.preferredReadReplica());
+ }
+
+ public static boolean isPreferredReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() != INVALID_PREFERRED_REPLICA_ID;
+ }
+
+ public static FetchResponseData.PartitionData partitionResponse(int partition, Errors error) {
+ return new FetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK);
+ }
+
+ /**
+ * cast the BaseRecords of PartitionData to Records. KRPC converts the byte array to MemoryRecords so this method
+ * never fail if the data is from KRPC.
+ *
+ * @param partition partition data
+ * @return Records or empty record if the records in PartitionData is null.
+ */
+ public static Records records(FetchResponseData.PartitionData partition) {
Review comment:
There is one place in this PR that we check for null when computing the records size, maybe we can use this utility function there.
##########
File path: core/src/test/scala/unit/kafka/server/FetchSessionTest.scala
##########
@@ -658,12 +732,19 @@ class FetchSessionTest {
// Full fetch context returns all partitions in the response
val context1 = fetchManager.newContext(JFetchMetadata.INITIAL, reqData, EMPTY_PART_LIST, isFollower = false)
assertEquals(classOf[FullFetchContext], context1.getClass)
- val respData = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[Records]]
- respData.put(tp1, new FetchResponse.PartitionData(Errors.NONE,
- 105, 105, 0, Optional.empty(), Collections.emptyList(), Optional.empty(), null))
- val divergingEpoch = Optional.of(new FetchResponseData.EpochEndOffset().setEpoch(3).setEndOffset(90))
- respData.put(tp2, new FetchResponse.PartitionData(Errors.NONE,
- 105, 105, 0, Optional.empty(), Collections.emptyList(), divergingEpoch, null))
+ val respData = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
+ respData.put(tp1, new FetchResponseData.PartitionData()
+ .setHighWatermark(105)
+ .setLastStableOffset(105)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(Collections.emptyList()))
Review comment:
Redundant.
##########
File path: jmh-benchmarks/src/main/java/org/apache/kafka/jmh/common/FetchResponseBenchmark.java
##########
@@ -78,19 +78,23 @@ public void setup() {
for (int topicIdx = 0; topicIdx < topicCount; topicIdx++) {
String topic = UUID.randomUUID().toString();
for (int partitionId = 0; partitionId < partitionCount; partitionId++) {
- FetchResponse.PartitionData<MemoryRecords> partitionData = new FetchResponse.PartitionData<>(
- Errors.NONE, 0, 0, 0, Optional.empty(), Collections.emptyList(), records);
+ FetchResponseData.PartitionData partitionData = new FetchResponseData.PartitionData()
+ .setPartitionIndex(partitionId)
+ .setLastStableOffset(0)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(Collections.emptyList())
Review comment:
Redundant.
##########
File path: core/src/test/scala/unit/kafka/server/FetchSessionTest.scala
##########
@@ -534,15 +588,21 @@ class FetchSessionTest {
Optional.empty()))
val session2context = fetchManager.newContext(JFetchMetadata.INITIAL, session1req, EMPTY_PART_LIST, false)
assertEquals(classOf[FullFetchContext], session2context.getClass)
- val session2RespData = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[Records]]
- session2RespData.put(new TopicPartition("foo", 0), new FetchResponse.PartitionData(
- Errors.NONE, 100, 100, 100, null, null))
- session2RespData.put(new TopicPartition("foo", 1), new FetchResponse.PartitionData(
- Errors.NONE, 10, 10, 10, null, null))
+ val session2RespData = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
+ session2RespData.put(new TopicPartition("foo", 0),
+ new FetchResponseData.PartitionData()
+ .setHighWatermark(100)
+ .setLastStableOffset(100)
+ .setLogStartOffset(100))
+ session2RespData.put(new TopicPartition("foo", 1),
+ new FetchResponseData.PartitionData()
+ .setHighWatermark(10)
Review comment:
Do we need to set the partition id here and other cases?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-790617559
> Remove unnecessary copying, if possible, in KafkaApis
https://issues.apache.org/jira/browse/KAFKA-12385
> Remove the lazily populated map in FetchResponse
https://issues.apache.org/jira/browse/KAFKA-12387
> I think you had an idea for removing FetchResponse.toMessage
ugh, I missed this one. opened: https://issues.apache.org/jira/browse/KAFKA-12410
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r551412139
##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
##########
@@ -733,11 +733,11 @@ void resetOffsetIfNeeded(TopicPartition partition, OffsetResetStrategy requested
}
private void resetOffsetsAsync(Map<TopicPartition, Long> partitionResetTimestamps) {
- Map<Node, Map<TopicPartition, ListOffsetsPartition>> timestampsToSearchByNode =
+ Map<Node, Map<TopicPartition, ListOffsetsRequestData.ListOffsetsPartition>> timestampsToSearchByNode =
groupListOffsetRequests(partitionResetTimestamps, new HashSet<>());
- for (Map.Entry<Node, Map<TopicPartition, ListOffsetsPartition>> entry : timestampsToSearchByNode.entrySet()) {
+ for (Map.Entry<Node, Map<TopicPartition, ListOffsetsRequestData.ListOffsetsPartition>> entry : timestampsToSearchByNode.entrySet()) {
Node node = entry.getKey();
- final Map<TopicPartition, ListOffsetsPartition> resetTimestamps = entry.getValue();
+ final Map<TopicPartition, ListOffsetsRequestData.ListOffsetsPartition> resetTimestamps = entry.getValue();
Review comment:
These changes seem unrelated?
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -59,136 +60,140 @@
*/
public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
- public static final long INVALID_HIGHWATERMARK = -1L;
+ public static FetchResponseData.FetchablePartitionResponse partitionResponse(Errors error) {
+ return new FetchResponseData.FetchablePartitionResponse()
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK)
+ .setLastStableOffset(FetchResponse.INVALID_LAST_STABLE_OFFSET)
+ .setLogStartOffset(FetchResponse.INVALID_LOG_START_OFFSET)
+ .setAbortedTransactions(null)
+ .setRecordSet(MemoryRecords.EMPTY)
+ .setPreferredReadReplica(FetchResponse.INVALID_PREFERRED_REPLICA_ID);
+ }
+
+ public static final long INVALID_HIGH_WATERMARK = -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 FetchResponseData data;
- private final LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap;
+ private final LinkedHashMap<TopicPartition, FetchResponseData.FetchablePartitionResponse> partitionData;
+ // lazily instantiate this field
+ private LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap = null;
Review comment:
Can we remove this altogether?
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -59,136 +60,140 @@
*/
public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
- public static final long INVALID_HIGHWATERMARK = -1L;
+ public static FetchResponseData.FetchablePartitionResponse partitionResponse(Errors error) {
+ return new FetchResponseData.FetchablePartitionResponse()
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK)
+ .setLastStableOffset(FetchResponse.INVALID_LAST_STABLE_OFFSET)
+ .setLogStartOffset(FetchResponse.INVALID_LOG_START_OFFSET)
+ .setAbortedTransactions(null)
+ .setRecordSet(MemoryRecords.EMPTY)
+ .setPreferredReadReplica(FetchResponse.INVALID_PREFERRED_REPLICA_ID);
+ }
+
+ public static final long INVALID_HIGH_WATERMARK = -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 FetchResponseData data;
- private final LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap;
+ private final LinkedHashMap<TopicPartition, FetchResponseData.FetchablePartitionResponse> partitionData;
+ // lazily instantiate this field
+ private LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap = null;
@Override
public FetchResponseData data() {
return data;
}
- public static final class AbortedTransaction {
- public final long producerId;
- public final long firstOffset;
+ /**
+ * From version 3 or later, the entries in `responseData` should be in the same order as the entries in
+ * `FetchRequest.fetchData`.
+ *
+ * @param error The top-level error code.
+ * @param responseData The fetched data grouped by partition.
+ * @param throttleTimeMs The time in milliseconds that the response was throttled
+ * @param sessionId The fetch session id.
+ */
+ public FetchResponse(Errors error,
+ LinkedHashMap<TopicPartition, PartitionData<T>> responseData,
+ int throttleTimeMs,
+ int sessionId) {
+ this(error, throttleTimeMs, sessionId, responseData.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey,
+ entry -> entry.getValue().partitionResponse, (o1, o2) -> {
+ throw new RuntimeException("this is impossible");
+ }, LinkedHashMap::new)));
+ }
- public AbortedTransaction(long producerId, long firstOffset) {
- this.producerId = producerId;
- this.firstOffset = firstOffset;
- }
+ public FetchResponse(Errors error,
+ int throttleTimeMs,
+ int sessionId,
+ LinkedHashMap<TopicPartition, FetchResponseData.FetchablePartitionResponse> responseData) {
+ super(ApiKeys.FETCH);
+ this.data = new FetchResponseData()
+ .setSessionId(sessionId)
+ .setErrorCode(error.code())
+ .setThrottleTimeMs(throttleTimeMs);
+ responseData.forEach((tp, tpData) -> data.responses().add(new FetchResponseData.FetchableTopicResponse()
+ .setTopic(tp.topic())
+ .setPartitionResponses(Collections.singletonList(tpData.setPartition(tp.partition())))));
+ this.partitionData = responseData;
+ }
- @Override
- public boolean equals(Object o) {
- if (this == o)
- return true;
- if (o == null || getClass() != o.getClass())
- return false;
+ public FetchResponse(FetchResponseData fetchResponseData) {
+ super(ApiKeys.FETCH);
+ this.data = fetchResponseData;
+ this.partitionData = new LinkedHashMap<>();
+ fetchResponseData.responses().forEach(topicResponse ->
+ topicResponse.partitionResponses().forEach(partitionResponse ->
+ partitionData.put(new TopicPartition(topicResponse.topic(), partitionResponse.partition()), partitionResponse))
+ );
+ }
- AbortedTransaction that = (AbortedTransaction) o;
+ public Errors error() {
+ return Errors.forCode(data.errorCode());
+ }
- return producerId == that.producerId && firstOffset == that.firstOffset;
+ public LinkedHashMap<TopicPartition, PartitionData<T>> responseData() {
+ if (responseDataMap == null) {
+ responseDataMap = new LinkedHashMap<>(partitionData.size());
+ partitionData.forEach((tp, d) -> responseDataMap.put(tp, new PartitionData<>(d)));
}
+ return responseDataMap;
+ }
- @Override
- public int hashCode() {
- int result = Long.hashCode(producerId);
- result = 31 * result + Long.hashCode(firstOffset);
- return result;
- }
+ @Override
+ public int throttleTimeMs() {
+ return data.throttleTimeMs();
+ }
- @Override
- public String toString() {
- return "(producerId=" + producerId + ", firstOffset=" + firstOffset + ")";
- }
+ public int sessionId() {
+ return data.sessionId();
+ }
- static AbortedTransaction fromMessage(FetchResponseData.AbortedTransaction abortedTransaction) {
- return new AbortedTransaction(abortedTransaction.producerId(), abortedTransaction.firstOffset());
- }
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ Map<Errors, Integer> errorCounts = new HashMap<>();
+ updateErrorCounts(errorCounts, error());
+ partitionData.values().forEach(response ->
+ updateErrorCounts(errorCounts, Errors.forCode(response.errorCode()))
+ );
+ return errorCounts;
}
- public static final class PartitionData<T extends BaseRecords> {
- private final FetchResponseData.FetchablePartitionResponse partitionResponse;
+ public static FetchResponse<MemoryRecords> parse(ByteBuffer buffer, short version) {
+ return new FetchResponse<>(new FetchResponseData(new ByteBufferAccessor(buffer), version));
+ }
- // Derived fields
- private final Optional<Integer> preferredReplica;
- private final List<AbortedTransaction> abortedTransactions;
- private final Errors error;
+ /**
+ * Convenience method to find the size of a response.
+ *
+ * @param version The version of the response to use.
+ * @param partIterator The partition iterator.
+ * @return The response size in bytes.
+ */
+ public static <T extends Records> int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ // Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
+ // use arbitrary values here without affecting the result.
+ LinkedHashMap<TopicPartition, PartitionData<T>> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
+ ObjectSerializationCache cache = new ObjectSerializationCache();
+ return 4 + new FetchResponse<>(Errors.NONE, data, 0, INVALID_SESSION_ID).data.size(cache, version);
+ }
- 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.preferredReadReplica())
- .filter(replicaId -> replicaId != INVALID_PREFERRED_REPLICA_ID);
-
- if (partitionResponse.abortedTransactions() == null) {
- this.abortedTransactions = null;
- } else {
- this.abortedTransactions = partitionResponse.abortedTransactions().stream()
- .map(AbortedTransaction::fromMessage)
- .collect(Collectors.toList());
- }
-
- this.error = Errors.forCode(partitionResponse.errorCode());
- }
+ @Override
+ public boolean shouldClientThrottle(short version) {
+ return version >= 8;
+ }
- public PartitionData(Errors error,
- long highWatermark,
- long lastStableOffset,
- long logStartOffset,
- Optional<Integer> preferredReadReplica,
- List<AbortedTransaction> abortedTransactions,
- Optional<FetchResponseData.EpochEndOffset> divergingEpoch,
- T records) {
- this.preferredReplica = preferredReadReplica;
- this.abortedTransactions = abortedTransactions;
- this.error = error;
-
- FetchResponseData.FetchablePartitionResponse partitionResponse =
- new FetchResponseData.FetchablePartitionResponse();
- partitionResponse.setErrorCode(error.code())
- .setHighWatermark(highWatermark)
- .setLastStableOffset(lastStableOffset)
- .setLogStartOffset(logStartOffset);
- if (abortedTransactions != null) {
- partitionResponse.setAbortedTransactions(abortedTransactions.stream().map(
- aborted -> new FetchResponseData.AbortedTransaction()
- .setProducerId(aborted.producerId)
- .setFirstOffset(aborted.firstOffset))
- .collect(Collectors.toList()));
- } else {
- partitionResponse.setAbortedTransactions(null);
- }
- partitionResponse.setPreferredReadReplica(preferredReadReplica.orElse(INVALID_PREFERRED_REPLICA_ID));
- partitionResponse.setRecordSet(records);
- divergingEpoch.ifPresent(partitionResponse::setDivergingEpoch);
- this.partitionResponse = partitionResponse;
- }
+ public static final class PartitionData<T extends BaseRecords> {
Review comment:
Do we need this class? What does it add over `FetchResponseData.FetchablePartitionRespons`?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584151716
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -57,238 +56,39 @@
* the fetch offset after the index lookup
* - {@link Errors#UNKNOWN_SERVER_ERROR} For any unexpected errors
*/
-public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
-
- public static final long INVALID_HIGHWATERMARK = -1L;
+public class FetchResponse extends AbstractResponse {
+ public static final long INVALID_HIGH_WATERMARK = -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 FetchResponseData data;
- private final LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap;
+ // we build responseData when needed.
+ private LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData = null;
Review comment:
Yeah, you can use a volatile field and synchronize on the assignment if still null. And then file a separate Jira to remove it from the class in a separate PR.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r585221282
##########
File path: core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala
##########
@@ -1144,8 +1143,14 @@ class AbstractFetcherThreadTest {
(Errors.NONE, records)
}
- (partition, new FetchData(error, leaderState.highWatermark, leaderState.highWatermark, leaderState.logStartOffset,
- Optional.empty[Integer], List.empty.asJava, divergingEpoch.asJava, records))
+ (partition, new FetchResponseData.PartitionData()
+ .setErrorCode(error.code)
+ .setHighWatermark(leaderState.highWatermark)
+ .setLastStableOffset(leaderState.highWatermark)
+ .setLogStartOffset(leaderState.logStartOffset)
+ .setAbortedTransactions(Collections.emptyList())
+ .setRecords(records)
+ .setDivergingEpoch(divergingEpoch.getOrElse(new FetchResponseData.EpochEndOffset)))
Review comment:
the type `FetchablePartitionResponse` is gone and the replacement `FetchResponseData.PartitionData` (generated data) can't accept `Optional` type.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584148817
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -57,238 +56,39 @@
* the fetch offset after the index lookup
* - {@link Errors#UNKNOWN_SERVER_ERROR} For any unexpected errors
*/
-public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
-
- public static final long INVALID_HIGHWATERMARK = -1L;
+public class FetchResponse extends AbstractResponse {
+ public static final long INVALID_HIGH_WATERMARK = -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 FetchResponseData data;
- private final LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap;
+ // we build responseData when needed.
+ private LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData = null;
Review comment:
It is used by `FetchSessionHandler` also. I prefer to rewrite that code by another PR to avoid big patch :(
It seems to me using synchronization block can resolve the thread issue. Also, I will file a jira as follow-up. WDYT?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r585691505
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -761,79 +754,85 @@ class KafkaApis(val requestChannel: RequestChannel,
// For fetch requests from clients, check if down-conversion is disabled for the particular partition
if (!fetchRequest.isFromFollower && !logConfig.forall(_.messageDownConversionEnable)) {
trace(s"Conversion to message format ${downConvertMagic.get} is disabled for partition $tp. Sending unsupported version response to $clientId.")
- errorResponse(Errors.UNSUPPORTED_VERSION)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_VERSION)
} else {
try {
trace(s"Down converting records from partition $tp to message format version $magic for fetch request from $clientId")
// Because down-conversion is extremely memory intensive, we want to try and delay the down-conversion as much
// 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.
- val error = maybeDownConvertStorageError(partitionData.error)
- 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))
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(new LazyDownConversionRecords(tp, unconvertedRecords, magic, fetchContext.getFetchOffset(tp).get, time))
+ .setPreferredReadReplica(partitionData.preferredReadReplica())
} catch {
case e: UnsupportedCompressionTypeException =>
trace("Received unsupported compression type error during down-conversion", e)
- errorResponse(Errors.UNSUPPORTED_COMPRESSION_TYPE)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_COMPRESSION_TYPE)
}
}
case None =>
- val error = maybeDownConvertStorageError(partitionData.error)
- new FetchResponse.PartitionData[BaseRecords](error,
- partitionData.highWatermark,
- partitionData.lastStableOffset,
- partitionData.logStartOffset,
- partitionData.preferredReadReplica,
- partitionData.abortedTransactions,
- partitionData.divergingEpoch,
- unconvertedRecords)
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(unconvertedRecords)
+ .setPreferredReadReplica(partitionData.preferredReadReplica)
+ .setDivergingEpoch(partitionData.divergingEpoch)
}
}
}
// the callback for process a fetch response, invoked before throttling
def processResponseCallback(responsePartitionData: Seq[(TopicPartition, FetchPartitionData)]): Unit = {
- val partitions = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[Records]]
+ val partitions = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
val reassigningPartitions = mutable.Set[TopicPartition]()
responsePartitionData.foreach { case (tp, data) =>
val abortedTransactions = data.abortedTransactions.map(_.asJava).orNull
val lastStableOffset = data.lastStableOffset.getOrElse(FetchResponse.INVALID_LAST_STABLE_OFFSET)
- if (data.isReassignmentFetch)
- reassigningPartitions.add(tp)
- val error = maybeDownConvertStorageError(data.error)
- partitions.put(tp, new FetchResponse.PartitionData(
- error,
- data.highWatermark,
- lastStableOffset,
- data.logStartOffset,
- data.preferredReadReplica.map(int2Integer).asJava,
- abortedTransactions,
- data.divergingEpoch.asJava,
- data.records))
+ if (data.isReassignmentFetch) reassigningPartitions.add(tp)
+ val partitionData = new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(data.error).code)
+ .setHighWatermark(data.highWatermark)
+ .setLastStableOffset(lastStableOffset)
+ .setLogStartOffset(data.logStartOffset)
+ .setAbortedTransactions(abortedTransactions)
+ .setRecords(data.records)
+ .setPreferredReadReplica(data.preferredReadReplica.getOrElse(FetchResponse.INVALID_PREFERRED_REPLICA_ID))
+ data.divergingEpoch.foreach(partitionData.setDivergingEpoch)
+ partitions.put(tp, partitionData)
}
erroneous.foreach { case (tp, data) => partitions.put(tp, data) }
- var unconvertedFetchResponse: FetchResponse[Records] = null
+ var unconvertedFetchResponse: FetchResponse = null
- def createResponse(throttleTimeMs: Int): FetchResponse[BaseRecords] = {
+ def createResponse(throttleTimeMs: Int): FetchResponse = {
// Down-convert messages for each partition if required
- val convertedData = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[BaseRecords]]
+ val convertedData = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
unconvertedFetchResponse.responseData.forEach { (tp, unconvertedPartitionData) =>
- if (unconvertedPartitionData.error != Errors.NONE)
+ val error = Errors.forCode(unconvertedPartitionData.errorCode)
+ if (error != Errors.NONE)
debug(s"Fetch request with correlation id ${request.header.correlationId} from client $clientId " +
- s"on partition $tp failed due to ${unconvertedPartitionData.error.exceptionName}")
+ s"on partition $tp failed due to ${error.exceptionName}")
convertedData.put(tp, maybeConvertFetchedData(tp, unconvertedPartitionData))
}
// Prepare fetch response from converted data
- val response = new FetchResponse(unconvertedFetchResponse.error, convertedData, throttleTimeMs,
- unconvertedFetchResponse.sessionId)
+ val response = FetchResponse.of(unconvertedFetchResponse.error, throttleTimeMs, unconvertedFetchResponse.sessionId, convertedData)
// record the bytes out metrics only when the response is being sent
response.responseData.forEach { (tp, data) =>
- brokerTopicStats.updateBytesOut(tp.topic, fetchRequest.isFromFollower, reassigningPartitions.contains(tp), data.records.sizeInBytes)
+ brokerTopicStats.updateBytesOut(tp.topic, fetchRequest.isFromFollower,
+ reassigningPartitions.contains(tp), if (data.records == null) 0 else data.records.sizeInBytes)
Review comment:
Ah, I thought the `size` method was in `FetchResponse`. Maybe we should add one? That would always be safe to call.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-762440344
@ijuma I add some perf results to description. In short, I don't observe significant regression.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584142186
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -57,238 +56,39 @@
* the fetch offset after the index lookup
* - {@link Errors#UNKNOWN_SERVER_ERROR} For any unexpected errors
*/
-public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
-
- public static final long INVALID_HIGHWATERMARK = -1L;
+public class FetchResponse extends AbstractResponse {
+ public static final long INVALID_HIGH_WATERMARK = -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 FetchResponseData data;
- private final LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap;
+ // we build responseData when needed.
+ private LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData = null;
Review comment:
I think I would move the method to a test utility so that tests can use that instead.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584222186
##########
File path: jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetcher/ReplicaFetcherThreadBenchmark.java
##########
@@ -174,8 +173,14 @@ public int sizeInBytes() {
return null;
}
};
- initialFetched.put(tp, new FetchResponse.PartitionData<>(Errors.NONE, 0, 0, 0,
- new LinkedList<>(), fetched));
+ initialFetched.put(tp, new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition())
+ .setErrorCode(Errors.NONE.code())
+ .setHighWatermark(0)
+ .setLastStableOffset(0)
+ .setLogStartOffset(0)
Review comment:
I will remove redundant setter `setHighWatermark(0)`. Other values are different from default value in KRPC.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r585656791
##########
File path: clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
##########
@@ -791,15 +791,17 @@ public void produceRequestGetErrorResponseTest() {
@Test
public void fetchResponseVersionTest() {
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> responseData = new LinkedHashMap<>();
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData = new LinkedHashMap<>();
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(), Collections.emptyList(), records));
-
- FetchResponse<MemoryRecords> v0Response = new FetchResponse<>(Errors.NONE, responseData, 0, INVALID_SESSION_ID);
- FetchResponse<MemoryRecords> v1Response = new FetchResponse<>(Errors.NONE, responseData, 10, INVALID_SESSION_ID);
+ responseData.put(new TopicPartition("test", 0),
+ new FetchResponseData.PartitionData()
Review comment:
Set partition id.
##########
File path: clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
##########
@@ -1146,38 +1160,45 @@ private FetchRequest createFetchRequest(int version) {
return FetchRequest.Builder.forConsumer(100, 100000, fetchData).setMaxBytes(1000).build((short) version);
}
- private FetchResponse<MemoryRecords> createFetchResponse(Errors error, int sessionId) {
- return new FetchResponse<>(error, new LinkedHashMap<>(), 25, sessionId);
+ private FetchResponse createFetchResponse(Errors error, int sessionId) {
+ return FetchResponse.of(error, 25, sessionId, new LinkedHashMap<>());
}
- private FetchResponse<MemoryRecords> createFetchResponse(int sessionId) {
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> responseData = new LinkedHashMap<>();
+ private FetchResponse createFetchResponse(int sessionId) {
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> 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(), Collections.emptyList(), records));
- List<FetchResponse.AbortedTransaction> 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));
- return new FetchResponse<>(Errors.NONE, responseData, 25, sessionId);
- }
-
- private FetchResponse<MemoryRecords> createFetchResponse(boolean includeAborted) {
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> responseData = new LinkedHashMap<>();
+ responseData.put(new TopicPartition("test", 0), new FetchResponseData.PartitionData()
+ .setHighWatermark(1000000)
Review comment:
Set partition id.
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -761,79 +754,85 @@ class KafkaApis(val requestChannel: RequestChannel,
// For fetch requests from clients, check if down-conversion is disabled for the particular partition
if (!fetchRequest.isFromFollower && !logConfig.forall(_.messageDownConversionEnable)) {
trace(s"Conversion to message format ${downConvertMagic.get} is disabled for partition $tp. Sending unsupported version response to $clientId.")
- errorResponse(Errors.UNSUPPORTED_VERSION)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_VERSION)
} else {
try {
trace(s"Down converting records from partition $tp to message format version $magic for fetch request from $clientId")
// Because down-conversion is extremely memory intensive, we want to try and delay the down-conversion as much
// 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.
- val error = maybeDownConvertStorageError(partitionData.error)
- 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))
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(new LazyDownConversionRecords(tp, unconvertedRecords, magic, fetchContext.getFetchOffset(tp).get, time))
+ .setPreferredReadReplica(partitionData.preferredReadReplica())
} catch {
case e: UnsupportedCompressionTypeException =>
trace("Received unsupported compression type error during down-conversion", e)
- errorResponse(Errors.UNSUPPORTED_COMPRESSION_TYPE)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_COMPRESSION_TYPE)
}
}
case None =>
- val error = maybeDownConvertStorageError(partitionData.error)
- new FetchResponse.PartitionData[BaseRecords](error,
- partitionData.highWatermark,
- partitionData.lastStableOffset,
- partitionData.logStartOffset,
- partitionData.preferredReadReplica,
- partitionData.abortedTransactions,
- partitionData.divergingEpoch,
- unconvertedRecords)
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(unconvertedRecords)
+ .setPreferredReadReplica(partitionData.preferredReadReplica)
+ .setDivergingEpoch(partitionData.divergingEpoch)
}
}
}
// the callback for process a fetch response, invoked before throttling
def processResponseCallback(responsePartitionData: Seq[(TopicPartition, FetchPartitionData)]): Unit = {
- val partitions = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[Records]]
+ val partitions = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
val reassigningPartitions = mutable.Set[TopicPartition]()
responsePartitionData.foreach { case (tp, data) =>
val abortedTransactions = data.abortedTransactions.map(_.asJava).orNull
val lastStableOffset = data.lastStableOffset.getOrElse(FetchResponse.INVALID_LAST_STABLE_OFFSET)
- if (data.isReassignmentFetch)
- reassigningPartitions.add(tp)
- val error = maybeDownConvertStorageError(data.error)
- partitions.put(tp, new FetchResponse.PartitionData(
- error,
- data.highWatermark,
- lastStableOffset,
- data.logStartOffset,
- data.preferredReadReplica.map(int2Integer).asJava,
- abortedTransactions,
- data.divergingEpoch.asJava,
- data.records))
+ if (data.isReassignmentFetch) reassigningPartitions.add(tp)
+ val partitionData = new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(data.error).code)
+ .setHighWatermark(data.highWatermark)
+ .setLastStableOffset(lastStableOffset)
+ .setLogStartOffset(data.logStartOffset)
+ .setAbortedTransactions(abortedTransactions)
+ .setRecords(data.records)
+ .setPreferredReadReplica(data.preferredReadReplica.getOrElse(FetchResponse.INVALID_PREFERRED_REPLICA_ID))
+ data.divergingEpoch.foreach(partitionData.setDivergingEpoch)
+ partitions.put(tp, partitionData)
}
erroneous.foreach { case (tp, data) => partitions.put(tp, data) }
- var unconvertedFetchResponse: FetchResponse[Records] = null
+ var unconvertedFetchResponse: FetchResponse = null
- def createResponse(throttleTimeMs: Int): FetchResponse[BaseRecords] = {
+ def createResponse(throttleTimeMs: Int): FetchResponse = {
// Down-convert messages for each partition if required
- val convertedData = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[BaseRecords]]
+ val convertedData = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
unconvertedFetchResponse.responseData.forEach { (tp, unconvertedPartitionData) =>
- if (unconvertedPartitionData.error != Errors.NONE)
+ val error = Errors.forCode(unconvertedPartitionData.errorCode)
+ if (error != Errors.NONE)
debug(s"Fetch request with correlation id ${request.header.correlationId} from client $clientId " +
- s"on partition $tp failed due to ${unconvertedPartitionData.error.exceptionName}")
+ s"on partition $tp failed due to ${error.exceptionName}")
convertedData.put(tp, maybeConvertFetchedData(tp, unconvertedPartitionData))
}
// Prepare fetch response from converted data
- val response = new FetchResponse(unconvertedFetchResponse.error, convertedData, throttleTimeMs,
- unconvertedFetchResponse.sessionId)
+ val response = FetchResponse.of(unconvertedFetchResponse.error, throttleTimeMs, unconvertedFetchResponse.sessionId, convertedData)
// record the bytes out metrics only when the response is being sent
response.responseData.forEach { (tp, data) =>
- brokerTopicStats.updateBytesOut(tp.topic, fetchRequest.isFromFollower, reassigningPartitions.contains(tp), data.records.sizeInBytes)
+ brokerTopicStats.updateBytesOut(tp.topic, fetchRequest.isFromFollower,
+ reassigningPartitions.contains(tp), if (data.records == null) 0 else data.records.sizeInBytes)
Review comment:
We can use the utility method you added to avoid the null check.
##########
File path: clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
##########
@@ -1146,38 +1160,45 @@ private FetchRequest createFetchRequest(int version) {
return FetchRequest.Builder.forConsumer(100, 100000, fetchData).setMaxBytes(1000).build((short) version);
}
- private FetchResponse<MemoryRecords> createFetchResponse(Errors error, int sessionId) {
- return new FetchResponse<>(error, new LinkedHashMap<>(), 25, sessionId);
+ private FetchResponse createFetchResponse(Errors error, int sessionId) {
+ return FetchResponse.of(error, 25, sessionId, new LinkedHashMap<>());
}
- private FetchResponse<MemoryRecords> createFetchResponse(int sessionId) {
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> responseData = new LinkedHashMap<>();
+ private FetchResponse createFetchResponse(int sessionId) {
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> 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(), Collections.emptyList(), records));
- List<FetchResponse.AbortedTransaction> 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));
- return new FetchResponse<>(Errors.NONE, responseData, 25, sessionId);
- }
-
- private FetchResponse<MemoryRecords> createFetchResponse(boolean includeAborted) {
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> responseData = new LinkedHashMap<>();
+ responseData.put(new TopicPartition("test", 0), new FetchResponseData.PartitionData()
+ .setHighWatermark(1000000)
+ .setLogStartOffset(0)
+ .setRecords(records));
+ List<FetchResponseData.AbortedTransaction> abortedTransactions = Collections.singletonList(
+ new FetchResponseData.AbortedTransaction().setProducerId(234L).setFirstOffset(999L));
+ responseData.put(new TopicPartition("test", 1), new FetchResponseData.PartitionData()
+ .setHighWatermark(1000000)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(abortedTransactions));
+ return FetchResponse.of(Errors.NONE, 25, sessionId, responseData);
+ }
+
+ private FetchResponse createFetchResponse(boolean includeAborted) {
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData = new LinkedHashMap<>();
MemoryRecords records = MemoryRecords.withRecords(CompressionType.NONE, new SimpleRecord("blah".getBytes()));
+ responseData.put(new TopicPartition("test", 0), new FetchResponseData.PartitionData()
+ .setHighWatermark(1000000)
Review comment:
Set partition id.
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -365,17 +126,98 @@ public int sessionId() {
* @param partIterator The partition iterator.
* @return The response size in bytes.
*/
- public static <T extends BaseRecords> int sizeOf(short version,
- Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ public static int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, FetchResponseData.PartitionData>> partIterator) {
// Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
// use arbitrary values here without affecting the result.
- FetchResponseData data = toMessage(0, Errors.NONE, partIterator, INVALID_SESSION_ID);
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
ObjectSerializationCache cache = new ObjectSerializationCache();
- return 4 + data.size(cache, version);
+ return 4 + FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, data).data.size(cache, version);
}
@Override
public boolean shouldClientThrottle(short version) {
return version >= 8;
}
-}
+
+ public static Optional<FetchResponseData.EpochEndOffset> divergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() < 0 ? Optional.empty()
+ : Optional.of(partitionResponse.divergingEpoch());
+ }
+
+ public static boolean isDivergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() >= 0;
+ }
+
+ public static Optional<Integer> preferredReadReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() == INVALID_PREFERRED_REPLICA_ID ? Optional.empty()
+ : Optional.of(partitionResponse.preferredReadReplica());
+ }
+
+ public static boolean isPreferredReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() != INVALID_PREFERRED_REPLICA_ID;
+ }
+
+ public static FetchResponseData.PartitionData partitionResponse(int partition, Errors error) {
+ return new FetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK);
+ }
+
+ /**
+ * Returns `partition.records` as `Records` (instead of `BaseRecords`). If `records` is `null`, returns `MemoryRecords.EMPTY`.
+ *
+ * If this response was deserialized after a fetch, this method should never fail. An example where this would
+ * fail is a down-converted response (e.g. LazyDownConversionRecords) on the broker (before it's serialized and
+ * sent on the wire).
+ *
+ * @param partition partition data
+ * @return Records or empty record if the records in PartitionData is null.
+ */
+ public static Records records(FetchResponseData.PartitionData partition) {
+ if (partition.records() == null) return MemoryRecords.EMPTY;
+ else if (partition.records() instanceof Records) return (Records) partition.records();
+ else throw new IllegalStateException("the record type is " + partition.records().getClass().getSimpleName() +
Review comment:
No `else` needed since we used `return` for both other cases. For the exception, I think we can just throw `ClassCastException` since `IllegalStateException` doesn't fit very well for this case. I would also make the message a bit more generic to avoid it going stale when we add more `Records` subtypes. For example:
```java
"The record type is " + partition.records().getClass().getSimpleName() + ", which is not a subtype of " +
Records.class.getSimpleName() + ". This method is only safe to call if the `FetchResponse` was
deserialized from bytes."
##########
File path: clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
##########
@@ -1146,38 +1160,45 @@ private FetchRequest createFetchRequest(int version) {
return FetchRequest.Builder.forConsumer(100, 100000, fetchData).setMaxBytes(1000).build((short) version);
}
- private FetchResponse<MemoryRecords> createFetchResponse(Errors error, int sessionId) {
- return new FetchResponse<>(error, new LinkedHashMap<>(), 25, sessionId);
+ private FetchResponse createFetchResponse(Errors error, int sessionId) {
+ return FetchResponse.of(error, 25, sessionId, new LinkedHashMap<>());
}
- private FetchResponse<MemoryRecords> createFetchResponse(int sessionId) {
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> responseData = new LinkedHashMap<>();
+ private FetchResponse createFetchResponse(int sessionId) {
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> 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(), Collections.emptyList(), records));
- List<FetchResponse.AbortedTransaction> 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));
- return new FetchResponse<>(Errors.NONE, responseData, 25, sessionId);
- }
-
- private FetchResponse<MemoryRecords> createFetchResponse(boolean includeAborted) {
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> responseData = new LinkedHashMap<>();
+ responseData.put(new TopicPartition("test", 0), new FetchResponseData.PartitionData()
+ .setHighWatermark(1000000)
+ .setLogStartOffset(0)
+ .setRecords(records));
+ List<FetchResponseData.AbortedTransaction> abortedTransactions = Collections.singletonList(
+ new FetchResponseData.AbortedTransaction().setProducerId(234L).setFirstOffset(999L));
+ responseData.put(new TopicPartition("test", 1), new FetchResponseData.PartitionData()
+ .setHighWatermark(1000000)
Review comment:
Set partition id.
##########
File path: clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
##########
@@ -1146,38 +1160,45 @@ private FetchRequest createFetchRequest(int version) {
return FetchRequest.Builder.forConsumer(100, 100000, fetchData).setMaxBytes(1000).build((short) version);
}
- private FetchResponse<MemoryRecords> createFetchResponse(Errors error, int sessionId) {
- return new FetchResponse<>(error, new LinkedHashMap<>(), 25, sessionId);
+ private FetchResponse createFetchResponse(Errors error, int sessionId) {
+ return FetchResponse.of(error, 25, sessionId, new LinkedHashMap<>());
}
- private FetchResponse<MemoryRecords> createFetchResponse(int sessionId) {
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> responseData = new LinkedHashMap<>();
+ private FetchResponse createFetchResponse(int sessionId) {
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> 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(), Collections.emptyList(), records));
- List<FetchResponse.AbortedTransaction> 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));
- return new FetchResponse<>(Errors.NONE, responseData, 25, sessionId);
- }
-
- private FetchResponse<MemoryRecords> createFetchResponse(boolean includeAborted) {
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> responseData = new LinkedHashMap<>();
+ responseData.put(new TopicPartition("test", 0), new FetchResponseData.PartitionData()
+ .setHighWatermark(1000000)
+ .setLogStartOffset(0)
+ .setRecords(records));
+ List<FetchResponseData.AbortedTransaction> abortedTransactions = Collections.singletonList(
+ new FetchResponseData.AbortedTransaction().setProducerId(234L).setFirstOffset(999L));
+ responseData.put(new TopicPartition("test", 1), new FetchResponseData.PartitionData()
+ .setHighWatermark(1000000)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(abortedTransactions));
+ return FetchResponse.of(Errors.NONE, 25, sessionId, responseData);
+ }
+
+ private FetchResponse createFetchResponse(boolean includeAborted) {
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData = new LinkedHashMap<>();
MemoryRecords records = MemoryRecords.withRecords(CompressionType.NONE, new SimpleRecord("blah".getBytes()));
+ responseData.put(new TopicPartition("test", 0), new FetchResponseData.PartitionData()
+ .setHighWatermark(1000000)
+ .setLogStartOffset(0)
+ .setRecords(records));
- responseData.put(new TopicPartition("test", 0), new FetchResponse.PartitionData<>(Errors.NONE,
- 1000000, FetchResponse.INVALID_LAST_STABLE_OFFSET, 0L, Optional.empty(), Collections.emptyList(), records));
-
- List<FetchResponse.AbortedTransaction> abortedTransactions = Collections.emptyList();
+ List<FetchResponseData.AbortedTransaction> abortedTransactions = Collections.emptyList();
if (includeAborted) {
abortedTransactions = Collections.singletonList(
- new FetchResponse.AbortedTransaction(234L, 999L));
+ new FetchResponseData.AbortedTransaction().setProducerId(234L).setFirstOffset(999L));
}
- responseData.put(new TopicPartition("test", 1), new FetchResponse.PartitionData<>(Errors.NONE,
- 1000000, FetchResponse.INVALID_LAST_STABLE_OFFSET, 0L, Optional.empty(), abortedTransactions, MemoryRecords.EMPTY));
+ responseData.put(new TopicPartition("test", 1), new FetchResponseData.PartitionData()
+ .setHighWatermark(1000000)
Review comment:
Set partition id.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584172530
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -365,17 +123,70 @@ public int sessionId() {
* @param partIterator The partition iterator.
* @return The response size in bytes.
*/
- public static <T extends BaseRecords> int sizeOf(short version,
- Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ public static int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, FetchResponseData.PartitionData>> partIterator) {
// Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
// use arbitrary values here without affecting the result.
- FetchResponseData data = toMessage(0, Errors.NONE, partIterator, INVALID_SESSION_ID);
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
ObjectSerializationCache cache = new ObjectSerializationCache();
- return 4 + data.size(cache, version);
+ return 4 + FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, data).data.size(cache, version);
}
@Override
public boolean shouldClientThrottle(short version) {
return version >= 8;
}
-}
+
+ public static Optional<FetchResponseData.EpochEndOffset> divergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() < 0 ? Optional.empty()
+ : Optional.of(partitionResponse.divergingEpoch());
+ }
+
+ public static boolean isDivergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() >= 0;
+ }
+
+ public static Optional<Integer> preferredReadReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() == INVALID_PREFERRED_REPLICA_ID ? Optional.empty()
+ : Optional.of(partitionResponse.preferredReadReplica());
+ }
+
+ public static boolean isPreferredReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() != INVALID_PREFERRED_REPLICA_ID;
+ }
+
+ public static FetchResponseData.PartitionData partitionResponse(int partition, Errors error) {
+ return new FetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK)
+ .setLastStableOffset(FetchResponse.INVALID_LAST_STABLE_OFFSET)
+ .setLogStartOffset(FetchResponse.INVALID_LOG_START_OFFSET)
+ .setAbortedTransactions(null)
+ .setRecords(MemoryRecords.EMPTY);
+ }
+
+ /**
+ * cast the BaseRecords of PartitionData to Records. This is used to eliminate duplicate code of type casting.
Review comment:
Does this ever fail? If so, it would be good to explain under which conditions it can fail. Also "This is used to eliminate duplicate code of type casting." seems a bit redundant.
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -304,58 +108,12 @@ public int sessionId() {
public Map<Errors, Integer> errorCounts() {
Map<Errors, Integer> errorCounts = new HashMap<>();
updateErrorCounts(errorCounts, error());
- responseDataMap.values().forEach(response ->
- updateErrorCounts(errorCounts, response.error())
- );
+ responseData().values().forEach(response -> updateErrorCounts(errorCounts, Errors.forCode(response.errorCode())));
Review comment:
Can we update this not to use `responseData`? Then we at least have the right behavior for the broker and we can fix the clients in the subsequent PR.
##########
File path: clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
##########
@@ -2377,14 +2378,19 @@ private ListOffsetsResponse listOffsetsResponse(Map<TopicPartition, Long> partit
builder.append(0L, ("key-" + i).getBytes(), ("value-" + i).getBytes());
records = builder.build();
}
- tpResponses.put(partition, new FetchResponse.PartitionData<>(
- Errors.NONE, highWatermark, FetchResponse.INVALID_LAST_STABLE_OFFSET,
- logStartOffset, null, records));
+ tpResponses.put(partition,
+ new FetchResponseData.PartitionData()
+ .setErrorCode(Errors.NONE.code())
+ .setHighWatermark(highWatermark)
+ .setLastStableOffset(FetchResponse.INVALID_LAST_STABLE_OFFSET)
+ .setLogStartOffset(logStartOffset)
+ .setAbortedTransactions(null)
+ .setRecords(records));
Review comment:
Are some of these redundant? (eg `setAbortedTransactions(null)`)
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -365,17 +123,70 @@ public int sessionId() {
* @param partIterator The partition iterator.
* @return The response size in bytes.
*/
- public static <T extends BaseRecords> int sizeOf(short version,
- Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ public static int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, FetchResponseData.PartitionData>> partIterator) {
// Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
// use arbitrary values here without affecting the result.
- FetchResponseData data = toMessage(0, Errors.NONE, partIterator, INVALID_SESSION_ID);
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
ObjectSerializationCache cache = new ObjectSerializationCache();
- return 4 + data.size(cache, version);
+ return 4 + FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, data).data.size(cache, version);
}
@Override
public boolean shouldClientThrottle(short version) {
return version >= 8;
}
-}
+
+ public static Optional<FetchResponseData.EpochEndOffset> divergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() < 0 ? Optional.empty()
+ : Optional.of(partitionResponse.divergingEpoch());
+ }
+
+ public static boolean isDivergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() >= 0;
+ }
+
+ public static Optional<Integer> preferredReadReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() == INVALID_PREFERRED_REPLICA_ID ? Optional.empty()
+ : Optional.of(partitionResponse.preferredReadReplica());
+ }
+
+ public static boolean isPreferredReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() != INVALID_PREFERRED_REPLICA_ID;
+ }
+
+ public static FetchResponseData.PartitionData partitionResponse(int partition, Errors error) {
+ return new FetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK)
+ .setLastStableOffset(FetchResponse.INVALID_LAST_STABLE_OFFSET)
+ .setLogStartOffset(FetchResponse.INVALID_LOG_START_OFFSET)
+ .setAbortedTransactions(null)
+ .setRecords(MemoryRecords.EMPTY);
Review comment:
Aren't many of these set automatically by the generated classes?
##########
File path: clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
##########
@@ -1270,13 +1271,24 @@ public void testFetchPositionAfterException() {
assertEquals(1, fetcher.sendFetches());
- Map<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> partitions = new LinkedHashMap<>();
- partitions.put(tp1, new FetchResponse.PartitionData<>(Errors.NONE, 100,
- FetchResponse.INVALID_LAST_STABLE_OFFSET, FetchResponse.INVALID_LOG_START_OFFSET, null, records));
- partitions.put(tp0, new FetchResponse.PartitionData<>(Errors.OFFSET_OUT_OF_RANGE, 100,
- FetchResponse.INVALID_LAST_STABLE_OFFSET, FetchResponse.INVALID_LOG_START_OFFSET, null, MemoryRecords.EMPTY));
- client.prepareResponse(new FetchResponse<>(Errors.NONE, new LinkedHashMap<>(partitions),
- 0, INVALID_SESSION_ID));
+
+ Map<TopicPartition, FetchResponseData.PartitionData> partitions = new LinkedHashMap<>();
+ partitions.put(tp1, new FetchResponseData.PartitionData()
+ .setErrorCode(Errors.NONE.code())
+ .setHighWatermark(100)
+ .setLastStableOffset(FetchResponse.INVALID_LAST_STABLE_OFFSET)
+ .setLogStartOffset(FetchResponse.INVALID_LOG_START_OFFSET)
+ .setAbortedTransactions(null)
+ .setRecords(records));
+ partitions.put(tp0, new FetchResponseData.PartitionData()
+ .setErrorCode(Errors.OFFSET_OUT_OF_RANGE.code())
+ .setHighWatermark(100)
+ .setLastStableOffset(FetchResponse.INVALID_LAST_STABLE_OFFSET)
+ .setLogStartOffset(FetchResponse.INVALID_LOG_START_OFFSET)
+ .setAbortedTransactions(null)
+ .setRecords(MemoryRecords.EMPTY));
Review comment:
Are some of these redundant? (eg `setAbortedTransactions(null)`). Other examples in the same file.
##########
File path: jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetcher/ReplicaFetcherThreadBenchmark.java
##########
@@ -174,8 +173,14 @@ public int sizeInBytes() {
return null;
}
};
- initialFetched.put(tp, new FetchResponse.PartitionData<>(Errors.NONE, 0, 0, 0,
- new LinkedList<>(), fetched));
+ initialFetched.put(tp, new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition())
+ .setErrorCode(Errors.NONE.code())
+ .setHighWatermark(0)
+ .setLastStableOffset(0)
+ .setLogStartOffset(0)
Review comment:
We can remove some redundant setters?
##########
File path: core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala
##########
@@ -963,9 +966,13 @@ class ReplicaFetcherThreadTest {
val records = MemoryRecords.withRecords(CompressionType.NONE,
new SimpleRecord(1000, "foo".getBytes(StandardCharsets.UTF_8)))
-
- val partitionData: thread.FetchData = new FetchResponse.PartitionData[Records](
- Errors.NONE, 0, 0, 0, Optional.empty(), Collections.emptyList(), records)
+ val partitionData: thread.FetchData = new FetchResponseData.PartitionData()
+ .setErrorCode(Errors.NONE.code)
+ .setHighWatermark(0)
+ .setLastStableOffset(0)
+ .setLogStartOffset(0)
Review comment:
We can remove some redundant setters?
##########
File path: core/src/main/scala/kafka/tools/ReplicaVerificationTool.scala
##########
@@ -403,25 +401,24 @@ private class ReplicaFetcher(name: String, sourceBroker: Node, topicPartitions:
debug("Issuing fetch request ")
- var fetchResponse: FetchResponse[MemoryRecords] = null
+ var fetchResponse: FetchResponse = null
try {
val clientResponse = fetchEndpoint.sendRequest(fetchRequestBuilder)
- fetchResponse = clientResponse.responseBody.asInstanceOf[FetchResponse[MemoryRecords]]
+ fetchResponse = clientResponse.responseBody.asInstanceOf[FetchResponse]
} catch {
case t: Throwable =>
if (!isRunning)
throw t
}
if (fetchResponse != null) {
- fetchResponse.responseData.forEach { (tp, partitionData) =>
- replicaBuffer.addFetchedData(tp, sourceBroker.id, partitionData)
- }
+ fetchResponse.data.responses().forEach(topicResponse =>
+ topicResponse.partitions().forEach(partitionResponse =>
Review comment:
Nit: remove `()` twice.
##########
File path: core/src/test/scala/unit/kafka/server/FetchSessionTest.scala
##########
@@ -155,13 +155,28 @@ class FetchSessionTest {
assertEquals(Optional.of(1), epochs1(tp1))
assertEquals(Optional.of(2), epochs1(tp2))
- val response = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[Records]]
- response.put(tp0, new FetchResponse.PartitionData(Errors.NONE, 100, 100,
- 100, null, null))
- response.put(tp1, new FetchResponse.PartitionData(
- Errors.NONE, 10, 10, 10, null, null))
- response.put(tp2, new FetchResponse.PartitionData(
- Errors.NONE, 5, 5, 5, null, null))
+ val response = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
+ response.put(tp0, new FetchResponseData.PartitionData()
+ .setErrorCode(Errors.NONE.code)
+ .setHighWatermark(100)
+ .setLastStableOffset(100)
+ .setLogStartOffset(100)
+ .setAbortedTransactions(null)
+ .setRecords(null))
Review comment:
We can remove some redundant setters?
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -761,76 +754,80 @@ class KafkaApis(val requestChannel: RequestChannel,
// For fetch requests from clients, check if down-conversion is disabled for the particular partition
if (!fetchRequest.isFromFollower && !logConfig.forall(_.messageDownConversionEnable)) {
trace(s"Conversion to message format ${downConvertMagic.get} is disabled for partition $tp. Sending unsupported version response to $clientId.")
- errorResponse(Errors.UNSUPPORTED_VERSION)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_VERSION)
} else {
try {
trace(s"Down converting records from partition $tp to message format version $magic for fetch request from $clientId")
// Because down-conversion is extremely memory intensive, we want to try and delay the down-conversion as much
// 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.
- val error = maybeDownConvertStorageError(partitionData.error)
- 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))
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(new LazyDownConversionRecords(tp, unconvertedRecords, magic, fetchContext.getFetchOffset(tp).get, time))
+ .setPreferredReadReplica(partitionData.preferredReadReplica())
Review comment:
Do we have to copy like this or can we mutate the response?
##########
File path: clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java
##########
@@ -150,22 +150,23 @@ private static void assertListEquals(List<TopicPartition> expected, List<TopicPa
private static final class RespEntry {
final TopicPartition part;
- final FetchResponse.PartitionData<MemoryRecords> data;
+ final FetchResponseData.PartitionData data;
RespEntry(String topic, int partition, long highWatermark, long lastStableOffset) {
this.part = new TopicPartition(topic, partition);
- this.data = new FetchResponse.PartitionData<>(
- Errors.NONE,
- highWatermark,
- lastStableOffset,
- 0,
- null,
- null);
+
+ this.data = new FetchResponseData.PartitionData()
+ .setErrorCode(Errors.NONE.code())
Review comment:
Nit: indenting seems excessive.
##########
File path: core/src/main/scala/kafka/server/AbstractFetcherThread.scala
##########
@@ -364,7 +361,7 @@ abstract class AbstractFetcherThread(name: String,
}
}
if (isTruncationOnFetchSupported) {
- partitionData.divergingEpoch.ifPresent { divergingEpoch =>
+ FetchResponse.divergingEpoch(partitionData).ifPresent { divergingEpoch =>
Review comment:
Nit: indenting seems wrong.
##########
File path: clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java
##########
@@ -150,22 +150,23 @@ private static void assertListEquals(List<TopicPartition> expected, List<TopicPa
private static final class RespEntry {
final TopicPartition part;
- final FetchResponse.PartitionData<MemoryRecords> data;
+ final FetchResponseData.PartitionData data;
RespEntry(String topic, int partition, long highWatermark, long lastStableOffset) {
this.part = new TopicPartition(topic, partition);
- this.data = new FetchResponse.PartitionData<>(
- Errors.NONE,
- highWatermark,
- lastStableOffset,
- 0,
- null,
- null);
+
+ this.data = new FetchResponseData.PartitionData()
+ .setErrorCode(Errors.NONE.code())
+ .setHighWatermark(highWatermark)
+ .setLastStableOffset(lastStableOffset)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(null)
+ .setRecords(null);
Review comment:
Are some of these redundant? (eg setAbortedTransactions(null))
##########
File path: core/src/main/scala/kafka/server/AbstractFetcherThread.scala
##########
@@ -734,7 +730,7 @@ abstract class AbstractFetcherThread(name: String,
Option(partitionStates.stateValue(topicPartition))
}
- protected def toMemoryRecords(records: Records): MemoryRecords = {
+ protected def toMemoryRecords(records: BaseRecords): MemoryRecords = {
Review comment:
Do we have to update the matching inside the method to handle other potential records types? Or do we want to avoid changing this method signature instead?
##########
File path: jmh-benchmarks/src/main/java/org/apache/kafka/jmh/common/FetchResponseBenchmark.java
##########
@@ -78,19 +78,25 @@ public void setup() {
for (int topicIdx = 0; topicIdx < topicCount; topicIdx++) {
String topic = UUID.randomUUID().toString();
for (int partitionId = 0; partitionId < partitionCount; partitionId++) {
- FetchResponse.PartitionData<MemoryRecords> partitionData = new FetchResponse.PartitionData<>(
- Errors.NONE, 0, 0, 0, Optional.empty(), Collections.emptyList(), records);
+ FetchResponseData.PartitionData partitionData = new FetchResponseData.PartitionData()
+ .setPartitionIndex(partitionId)
+ .setErrorCode(Errors.NONE.code())
+ .setHighWatermark(0)
+ .setLastStableOffset(0)
+ .setLogStartOffset(0)
Review comment:
We can remove some redundant setters?
##########
File path: core/src/main/scala/kafka/server/AbstractFetcherThread.scala
##########
@@ -340,7 +336,8 @@ abstract class AbstractFetcherThread(name: String,
// the current offset is the same as the offset requested.
val fetchPartitionData = sessionPartitions.get(topicPartition)
if (fetchPartitionData != null && fetchPartitionData.fetchOffset == currentFetchState.fetchOffset && currentFetchState.isReadyForFetch) {
- partitionData.error match {
+ val partitionError = Errors.forCode(partitionData.errorCode)
Review comment:
Not clear why we need this val. Seems like we can introduce a variable in the `case _` instead.
##########
File path: jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetchsession/FetchSessionBenchmark.java
##########
@@ -70,24 +70,25 @@ public void setUp() {
handler = new FetchSessionHandler(LOG_CONTEXT, 1);
FetchSessionHandler.Builder builder = handler.newBuilder();
- LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> respMap = new LinkedHashMap<>();
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> respMap = new LinkedHashMap<>();
for (int i = 0; i < partitionCount; i++) {
TopicPartition tp = new TopicPartition("foo", i);
FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData(0, 0, 200,
Optional.empty());
fetches.put(tp, partitionData);
builder.add(tp, partitionData);
- respMap.put(tp, new FetchResponse.PartitionData<>(
- Errors.NONE,
- 0L,
- 0L,
- 0,
- null,
- null));
+ respMap.put(tp, new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition())
+ .setErrorCode(Errors.NONE.code())
+ .setHighWatermark(0)
+ .setLastStableOffset(0)
+ .setLogStartOffset(0)
+ .setAbortedTransactions(null)
+ .setRecords(null));
Review comment:
We can remove some redundant set calls?
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -761,76 +754,80 @@ class KafkaApis(val requestChannel: RequestChannel,
// For fetch requests from clients, check if down-conversion is disabled for the particular partition
if (!fetchRequest.isFromFollower && !logConfig.forall(_.messageDownConversionEnable)) {
trace(s"Conversion to message format ${downConvertMagic.get} is disabled for partition $tp. Sending unsupported version response to $clientId.")
- errorResponse(Errors.UNSUPPORTED_VERSION)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_VERSION)
} else {
try {
trace(s"Down converting records from partition $tp to message format version $magic for fetch request from $clientId")
// Because down-conversion is extremely memory intensive, we want to try and delay the down-conversion as much
// 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.
- val error = maybeDownConvertStorageError(partitionData.error)
- 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))
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(new LazyDownConversionRecords(tp, unconvertedRecords, magic, fetchContext.getFetchOffset(tp).get, time))
+ .setPreferredReadReplica(partitionData.preferredReadReplica())
} catch {
case e: UnsupportedCompressionTypeException =>
trace("Received unsupported compression type error during down-conversion", e)
- errorResponse(Errors.UNSUPPORTED_COMPRESSION_TYPE)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_COMPRESSION_TYPE)
}
}
case None =>
- val error = maybeDownConvertStorageError(partitionData.error)
- new FetchResponse.PartitionData[BaseRecords](error,
- partitionData.highWatermark,
- partitionData.lastStableOffset,
- partitionData.logStartOffset,
- partitionData.preferredReadReplica,
- partitionData.abortedTransactions,
- partitionData.divergingEpoch,
- unconvertedRecords)
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(unconvertedRecords)
+ .setPreferredReadReplica(partitionData.preferredReadReplica)
+ .setDivergingEpoch(partitionData.divergingEpoch)
Review comment:
Similar question, is the copy required?
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -761,76 +754,80 @@ class KafkaApis(val requestChannel: RequestChannel,
// For fetch requests from clients, check if down-conversion is disabled for the particular partition
if (!fetchRequest.isFromFollower && !logConfig.forall(_.messageDownConversionEnable)) {
trace(s"Conversion to message format ${downConvertMagic.get} is disabled for partition $tp. Sending unsupported version response to $clientId.")
- errorResponse(Errors.UNSUPPORTED_VERSION)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_VERSION)
} else {
try {
trace(s"Down converting records from partition $tp to message format version $magic for fetch request from $clientId")
// Because down-conversion is extremely memory intensive, we want to try and delay the down-conversion as much
// 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.
- val error = maybeDownConvertStorageError(partitionData.error)
- 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))
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(new LazyDownConversionRecords(tp, unconvertedRecords, magic, fetchContext.getFetchOffset(tp).get, time))
+ .setPreferredReadReplica(partitionData.preferredReadReplica())
} catch {
case e: UnsupportedCompressionTypeException =>
trace("Received unsupported compression type error during down-conversion", e)
- errorResponse(Errors.UNSUPPORTED_COMPRESSION_TYPE)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_COMPRESSION_TYPE)
}
}
case None =>
- val error = maybeDownConvertStorageError(partitionData.error)
- new FetchResponse.PartitionData[BaseRecords](error,
- partitionData.highWatermark,
- partitionData.lastStableOffset,
- partitionData.logStartOffset,
- partitionData.preferredReadReplica,
- partitionData.abortedTransactions,
- partitionData.divergingEpoch,
- unconvertedRecords)
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(unconvertedRecords)
+ .setPreferredReadReplica(partitionData.preferredReadReplica)
+ .setDivergingEpoch(partitionData.divergingEpoch)
}
}
}
// the callback for process a fetch response, invoked before throttling
def processResponseCallback(responsePartitionData: Seq[(TopicPartition, FetchPartitionData)]): Unit = {
- val partitions = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[Records]]
+ val partitions = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
val reassigningPartitions = mutable.Set[TopicPartition]()
responsePartitionData.foreach { case (tp, data) =>
val abortedTransactions = data.abortedTransactions.map(_.asJava).orNull
val lastStableOffset = data.lastStableOffset.getOrElse(FetchResponse.INVALID_LAST_STABLE_OFFSET)
- if (data.isReassignmentFetch)
- reassigningPartitions.add(tp)
- val error = maybeDownConvertStorageError(data.error)
- partitions.put(tp, new FetchResponse.PartitionData(
- error,
- data.highWatermark,
- lastStableOffset,
- data.logStartOffset,
- data.preferredReadReplica.map(int2Integer).asJava,
- abortedTransactions,
- data.divergingEpoch.asJava,
- data.records))
+ if (data.isReassignmentFetch) reassigningPartitions.add(tp)
+ partitions.put(tp, new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(data.error).code)
+ .setHighWatermark(data.highWatermark)
+ .setLastStableOffset(lastStableOffset)
+ .setLogStartOffset(data.logStartOffset)
+ .setAbortedTransactions(abortedTransactions)
+ .setRecords(data.records)
+ .setPreferredReadReplica(data.preferredReadReplica.getOrElse(FetchResponse.INVALID_PREFERRED_REPLICA_ID))
+ .setDivergingEpoch(data.divergingEpoch.getOrElse(new FetchResponseData.EpochEndOffset)))
Review comment:
Similar question, is the copy required?
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -304,58 +115,12 @@ public int sessionId() {
public Map<Errors, Integer> errorCounts() {
Map<Errors, Integer> errorCounts = new HashMap<>();
updateErrorCounts(errorCounts, error());
- responseDataMap.values().forEach(response ->
- updateErrorCounts(errorCounts, response.error())
- );
+ dataByTopicPartition.values().forEach(response -> updateErrorCounts(errorCounts, Errors.forCode(response.errorCode())));
return errorCounts;
}
- public static FetchResponse<MemoryRecords> parse(ByteBuffer buffer, short version) {
- return new FetchResponse<>(new FetchResponseData(new ByteBufferAccessor(buffer), version));
- }
-
- @SuppressWarnings("unchecked")
- private static <T extends BaseRecords> LinkedHashMap<TopicPartition, PartitionData<T>> toResponseDataMap(
- FetchResponseData message) {
- LinkedHashMap<TopicPartition, PartitionData<T>> responseMap = new LinkedHashMap<>();
- message.responses().forEach(topicResponse -> {
- topicResponse.partitionResponses().forEach(partitionResponse -> {
- TopicPartition tp = new TopicPartition(topicResponse.topic(), partitionResponse.partition());
- PartitionData<T> partitionData = new PartitionData<>(partitionResponse);
- responseMap.put(tp, partitionData);
- });
- });
- return responseMap;
- }
-
- private static <T extends BaseRecords> FetchResponseData toMessage(int throttleTimeMs, Errors error,
Review comment:
Can you clarify what you mean here?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r581088681
##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
##########
@@ -290,7 +291,7 @@ public void onSuccess(ClientResponse resp) {
Set<TopicPartition> partitions = new HashSet<>(response.responseData().keySet());
FetchResponseMetricAggregator metricAggregator = new FetchResponseMetricAggregator(sensors, partitions);
- for (Map.Entry<TopicPartition, FetchResponse.PartitionData<Records>> entry : response.responseData().entrySet()) {
+ for (Map.Entry<TopicPartition, FetchResponseData.FetchablePartitionResponse> entry : response.responseData().entrySet()) {
Review comment:
sure. I reuse the name `PartitionData`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584224693
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -761,76 +754,80 @@ class KafkaApis(val requestChannel: RequestChannel,
// For fetch requests from clients, check if down-conversion is disabled for the particular partition
if (!fetchRequest.isFromFollower && !logConfig.forall(_.messageDownConversionEnable)) {
trace(s"Conversion to message format ${downConvertMagic.get} is disabled for partition $tp. Sending unsupported version response to $clientId.")
- errorResponse(Errors.UNSUPPORTED_VERSION)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_VERSION)
} else {
try {
trace(s"Down converting records from partition $tp to message format version $magic for fetch request from $clientId")
// Because down-conversion is extremely memory intensive, we want to try and delay the down-conversion as much
// 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.
- val error = maybeDownConvertStorageError(partitionData.error)
- 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))
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(new LazyDownConversionRecords(tp, unconvertedRecords, magic, fetchContext.getFetchOffset(tp).get, time))
+ .setPreferredReadReplica(partitionData.preferredReadReplica())
Review comment:
It created copy before so I don't change the behavior in this PR. We can investigate it in separate PR :)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584221661
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -365,17 +123,70 @@ public int sessionId() {
* @param partIterator The partition iterator.
* @return The response size in bytes.
*/
- public static <T extends BaseRecords> int sizeOf(short version,
- Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ public static int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, FetchResponseData.PartitionData>> partIterator) {
// Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
// use arbitrary values here without affecting the result.
- FetchResponseData data = toMessage(0, Errors.NONE, partIterator, INVALID_SESSION_ID);
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
ObjectSerializationCache cache = new ObjectSerializationCache();
- return 4 + data.size(cache, version);
+ return 4 + FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, data).data.size(cache, version);
}
@Override
public boolean shouldClientThrottle(short version) {
return version >= 8;
}
-}
+
+ public static Optional<FetchResponseData.EpochEndOffset> divergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() < 0 ? Optional.empty()
+ : Optional.of(partitionResponse.divergingEpoch());
+ }
+
+ public static boolean isDivergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() >= 0;
+ }
+
+ public static Optional<Integer> preferredReadReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() == INVALID_PREFERRED_REPLICA_ID ? Optional.empty()
+ : Optional.of(partitionResponse.preferredReadReplica());
+ }
+
+ public static boolean isPreferredReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() != INVALID_PREFERRED_REPLICA_ID;
+ }
+
+ public static FetchResponseData.PartitionData partitionResponse(int partition, Errors error) {
+ return new FetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK)
+ .setLastStableOffset(FetchResponse.INVALID_LAST_STABLE_OFFSET)
+ .setLogStartOffset(FetchResponse.INVALID_LOG_START_OFFSET)
+ .setAbortedTransactions(null)
+ .setRecords(MemoryRecords.EMPTY);
Review comment:
you are right. Except for `HighWatermark`, other args have default value. remove duplicate assignment!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r581131032
##########
File path: clients/src/main/resources/common/message/FetchResponse.json
##########
@@ -53,9 +53,9 @@
"about": "The response topics.", "fields": [
{ "name": "Topic", "type": "string", "versions": "0+", "entityType": "topicName",
"about": "The topic name." },
- { "name": "PartitionResponses", "type": "[]FetchablePartitionResponse", "versions": "0+",
+ { "name": "Partitions", "type": "[]PartitionData", "versions": "0+",
"about": "The topic partitions.", "fields": [
- { "name": "Partition", "type": "int32", "versions": "0+",
+ { "name": "Index", "type": "int32", "versions": "0+",
Review comment:
> We typically call this PartitionIndex in other requests/responses. Is that right?
You are right!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-759464091
fix conflicting files again
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-794041782
> So is the plan to merge #10269 first or after #9944?
yep. #10269 is in progress (tweak performance). I will take a look at #9944 :)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r552043767
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -59,136 +60,140 @@
*/
public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
- public static final long INVALID_HIGHWATERMARK = -1L;
+ public static FetchResponseData.FetchablePartitionResponse partitionResponse(Errors error) {
+ return new FetchResponseData.FetchablePartitionResponse()
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK)
+ .setLastStableOffset(FetchResponse.INVALID_LAST_STABLE_OFFSET)
+ .setLogStartOffset(FetchResponse.INVALID_LOG_START_OFFSET)
+ .setAbortedTransactions(null)
+ .setRecordSet(MemoryRecords.EMPTY)
+ .setPreferredReadReplica(FetchResponse.INVALID_PREFERRED_REPLICA_ID);
+ }
+
+ public static final long INVALID_HIGH_WATERMARK = -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 FetchResponseData data;
- private final LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap;
+ private final LinkedHashMap<TopicPartition, FetchResponseData.FetchablePartitionResponse> partitionData;
+ // lazily instantiate this field
+ private LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap = null;
Review comment:
I understand. It's a bit hard to understand the end state for this class as it stands. We've taken a few intermediate steps to this point. Can we just go to the end state now?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r552053003
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -59,136 +60,140 @@
*/
public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
- public static final long INVALID_HIGHWATERMARK = -1L;
+ public static FetchResponseData.FetchablePartitionResponse partitionResponse(Errors error) {
+ return new FetchResponseData.FetchablePartitionResponse()
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK)
+ .setLastStableOffset(FetchResponse.INVALID_LAST_STABLE_OFFSET)
+ .setLogStartOffset(FetchResponse.INVALID_LOG_START_OFFSET)
+ .setAbortedTransactions(null)
+ .setRecordSet(MemoryRecords.EMPTY)
+ .setPreferredReadReplica(FetchResponse.INVALID_PREFERRED_REPLICA_ID);
+ }
+
+ public static final long INVALID_HIGH_WATERMARK = -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 FetchResponseData data;
- private final LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap;
+ private final LinkedHashMap<TopicPartition, FetchResponseData.FetchablePartitionResponse> partitionData;
+ // lazily instantiate this field
+ private LinkedHashMap<TopicPartition, PartitionData<T>> responseDataMap = null;
Review comment:
You are right. Will address it in next commit.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-790491007
```
Build / JDK 11 / org.apache.kafka.connect.integration.BlockingConnectorTest.testWorkerRestartWithBlockInConnectorStop
```
unrelated flaky
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584221805
##########
File path: core/src/main/scala/kafka/server/AbstractFetcherThread.scala
##########
@@ -340,7 +336,8 @@ abstract class AbstractFetcherThread(name: String,
// the current offset is the same as the offset requested.
val fetchPartitionData = sessionPartitions.get(topicPartition)
if (fetchPartitionData != null && fetchPartitionData.fetchOffset == currentFetchState.fetchOffset && currentFetchState.isReadyForFetch) {
- partitionData.error match {
+ val partitionError = Errors.forCode(partitionData.errorCode)
Review comment:
Good catch! updated
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jolshan commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
jolshan commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-794038027
@chia7712 sounds good. So is the plan to merge https://github.com/apache/kafka/pull/10269 first or after https://github.com/apache/kafka/pull/9944? Also thanks for making these changes. I noticed a lot of this when working on https://github.com/apache/kafka/pull/9944
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-794141685
Does the refactoring simplify #9944 or not?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r585660962
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -365,17 +126,98 @@ public int sessionId() {
* @param partIterator The partition iterator.
* @return The response size in bytes.
*/
- public static <T extends BaseRecords> int sizeOf(short version,
- Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ public static int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, FetchResponseData.PartitionData>> partIterator) {
// Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
// use arbitrary values here without affecting the result.
- FetchResponseData data = toMessage(0, Errors.NONE, partIterator, INVALID_SESSION_ID);
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
ObjectSerializationCache cache = new ObjectSerializationCache();
- return 4 + data.size(cache, version);
+ return 4 + FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, data).data.size(cache, version);
}
@Override
public boolean shouldClientThrottle(short version) {
return version >= 8;
}
-}
+
+ public static Optional<FetchResponseData.EpochEndOffset> divergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() < 0 ? Optional.empty()
+ : Optional.of(partitionResponse.divergingEpoch());
+ }
+
+ public static boolean isDivergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() >= 0;
+ }
+
+ public static Optional<Integer> preferredReadReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() == INVALID_PREFERRED_REPLICA_ID ? Optional.empty()
+ : Optional.of(partitionResponse.preferredReadReplica());
+ }
+
+ public static boolean isPreferredReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() != INVALID_PREFERRED_REPLICA_ID;
+ }
+
+ public static FetchResponseData.PartitionData partitionResponse(int partition, Errors error) {
+ return new FetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK);
+ }
+
+ /**
+ * Returns `partition.records` as `Records` (instead of `BaseRecords`). If `records` is `null`, returns `MemoryRecords.EMPTY`.
+ *
+ * If this response was deserialized after a fetch, this method should never fail. An example where this would
+ * fail is a down-converted response (e.g. LazyDownConversionRecords) on the broker (before it's serialized and
+ * sent on the wire).
+ *
+ * @param partition partition data
+ * @return Records or empty record if the records in PartitionData is null.
+ */
+ public static Records records(FetchResponseData.PartitionData partition) {
+ if (partition.records() == null) return MemoryRecords.EMPTY;
+ else if (partition.records() instanceof Records) return (Records) partition.records();
+ else throw new IllegalStateException("the record type is " + partition.records().getClass().getSimpleName() +
Review comment:
One more thing, let's call this `recordsOrFail` to make it clear that the operation is not necessarily safe.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r581128378
##########
File path: clients/src/main/resources/common/message/FetchResponse.json
##########
@@ -53,9 +53,9 @@
"about": "The response topics.", "fields": [
{ "name": "Topic", "type": "string", "versions": "0+", "entityType": "topicName",
"about": "The topic name." },
- { "name": "PartitionResponses", "type": "[]FetchablePartitionResponse", "versions": "0+",
+ { "name": "Partitions", "type": "[]PartitionData", "versions": "0+",
"about": "The topic partitions.", "fields": [
- { "name": "Partition", "type": "int32", "versions": "0+",
+ { "name": "Index", "type": "int32", "versions": "0+",
Review comment:
We typically call this `PartitionIndex` in other requests/responses. Is that right?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jolshan commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
jolshan commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-794195828
@ijuma on first glance, I'm not sure if it simplifies. But it will introduce changes. And these are likely changes we want to have eventually.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584222409
##########
File path: jmh-benchmarks/src/main/java/org/apache/kafka/jmh/common/FetchResponseBenchmark.java
##########
@@ -78,19 +78,25 @@ public void setup() {
for (int topicIdx = 0; topicIdx < topicCount; topicIdx++) {
String topic = UUID.randomUUID().toString();
for (int partitionId = 0; partitionId < partitionCount; partitionId++) {
- FetchResponse.PartitionData<MemoryRecords> partitionData = new FetchResponse.PartitionData<>(
- Errors.NONE, 0, 0, 0, Optional.empty(), Collections.emptyList(), records);
+ FetchResponseData.PartitionData partitionData = new FetchResponseData.PartitionData()
+ .setPartitionIndex(partitionId)
+ .setErrorCode(Errors.NONE.code())
+ .setHighWatermark(0)
+ .setLastStableOffset(0)
+ .setLogStartOffset(0)
Review comment:
updated
##########
File path: core/src/test/scala/unit/kafka/server/FetchSessionTest.scala
##########
@@ -155,13 +155,28 @@ class FetchSessionTest {
assertEquals(Optional.of(1), epochs1(tp1))
assertEquals(Optional.of(2), epochs1(tp2))
- val response = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[Records]]
- response.put(tp0, new FetchResponse.PartitionData(Errors.NONE, 100, 100,
- 100, null, null))
- response.put(tp1, new FetchResponse.PartitionData(
- Errors.NONE, 10, 10, 10, null, null))
- response.put(tp2, new FetchResponse.PartitionData(
- Errors.NONE, 5, 5, 5, null, null))
+ val response = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
+ response.put(tp0, new FetchResponseData.PartitionData()
+ .setErrorCode(Errors.NONE.code)
+ .setHighWatermark(100)
+ .setLastStableOffset(100)
+ .setLogStartOffset(100)
+ .setAbortedTransactions(null)
+ .setRecords(null))
Review comment:
updated
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r587216418
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -365,17 +126,105 @@ public int sessionId() {
* @param partIterator The partition iterator.
* @return The response size in bytes.
*/
- public static <T extends BaseRecords> int sizeOf(short version,
- Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ public static int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, FetchResponseData.PartitionData>> partIterator) {
// Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
// use arbitrary values here without affecting the result.
- FetchResponseData data = toMessage(0, Errors.NONE, partIterator, INVALID_SESSION_ID);
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
Review comment:
nice question. will revert to previous code :)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r585224204
##########
File path: core/src/test/scala/unit/kafka/server/FetchSessionTest.scala
##########
@@ -534,15 +588,21 @@ class FetchSessionTest {
Optional.empty()))
val session2context = fetchManager.newContext(JFetchMetadata.INITIAL, session1req, EMPTY_PART_LIST, false)
assertEquals(classOf[FullFetchContext], session2context.getClass)
- val session2RespData = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[Records]]
- session2RespData.put(new TopicPartition("foo", 0), new FetchResponse.PartitionData(
- Errors.NONE, 100, 100, 100, null, null))
- session2RespData.put(new TopicPartition("foo", 1), new FetchResponse.PartitionData(
- Errors.NONE, 10, 10, 10, null, null))
+ val session2RespData = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
+ session2RespData.put(new TopicPartition("foo", 0),
+ new FetchResponseData.PartitionData()
+ .setHighWatermark(100)
+ .setLastStableOffset(100)
+ .setLogStartOffset(100))
+ session2RespData.put(new TopicPartition("foo", 1),
+ new FetchResponseData.PartitionData()
+ .setHighWatermark(10)
Review comment:
good catch. will fix all similar issues in next commit.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jolshan edited a comment on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
jolshan edited a comment on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-794336035
@chia7712 I see. Thanks for the response. I'm not sure if https://github.com/apache/kafka/pull/9944 will be ready after fixing conflicts. There were still a few critical tests failing. I can try to fix those but still figuring out the timeline. I can go ahead and start with the conflicts though.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jolshan commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
jolshan commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-793082596
@chia7712 I noticed that all of these JIRAs are listed as open, but they reference this PR. Are there further changes to be made? Asking because I'm working on similar code for https://github.com/apache/kafka/pull/9944 and want to minimize merge conflicts as best as possible.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r552403231
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -304,58 +115,12 @@ public int sessionId() {
public Map<Errors, Integer> errorCounts() {
Map<Errors, Integer> errorCounts = new HashMap<>();
updateErrorCounts(errorCounts, error());
- responseDataMap.values().forEach(response ->
- updateErrorCounts(errorCounts, response.error())
- );
+ dataByTopicPartition.values().forEach(response -> updateErrorCounts(errorCounts, Errors.forCode(response.errorCode())));
return errorCounts;
}
- public static FetchResponse<MemoryRecords> parse(ByteBuffer buffer, short version) {
- return new FetchResponse<>(new FetchResponseData(new ByteBufferAccessor(buffer), version));
- }
-
- @SuppressWarnings("unchecked")
- private static <T extends BaseRecords> LinkedHashMap<TopicPartition, PartitionData<T>> toResponseDataMap(
- FetchResponseData message) {
- LinkedHashMap<TopicPartition, PartitionData<T>> responseMap = new LinkedHashMap<>();
- message.responses().forEach(topicResponse -> {
- topicResponse.partitionResponses().forEach(partitionResponse -> {
- TopicPartition tp = new TopicPartition(topicResponse.topic(), partitionResponse.partition());
- PartitionData<T> partitionData = new PartitionData<>(partitionResponse);
- responseMap.put(tp, partitionData);
- });
- });
- return responseMap;
- }
-
- private static <T extends BaseRecords> FetchResponseData toMessage(int throttleTimeMs, Errors error,
Review comment:
I don't batch the partitions again in this PR as it create a new ```FetchResponseData```.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r584221036
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -365,17 +123,70 @@ public int sessionId() {
* @param partIterator The partition iterator.
* @return The response size in bytes.
*/
- public static <T extends BaseRecords> int sizeOf(short version,
- Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator) {
+ public static int sizeOf(short version,
+ Iterator<Map.Entry<TopicPartition, FetchResponseData.PartitionData>> partIterator) {
// Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
// use arbitrary values here without affecting the result.
- FetchResponseData data = toMessage(0, Errors.NONE, partIterator, INVALID_SESSION_ID);
+ LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> data = new LinkedHashMap<>();
+ partIterator.forEachRemaining(entry -> data.put(entry.getKey(), entry.getValue()));
ObjectSerializationCache cache = new ObjectSerializationCache();
- return 4 + data.size(cache, version);
+ return 4 + FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, data).data.size(cache, version);
}
@Override
public boolean shouldClientThrottle(short version) {
return version >= 8;
}
-}
+
+ public static Optional<FetchResponseData.EpochEndOffset> divergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() < 0 ? Optional.empty()
+ : Optional.of(partitionResponse.divergingEpoch());
+ }
+
+ public static boolean isDivergingEpoch(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.divergingEpoch().epoch() >= 0;
+ }
+
+ public static Optional<Integer> preferredReadReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() == INVALID_PREFERRED_REPLICA_ID ? Optional.empty()
+ : Optional.of(partitionResponse.preferredReadReplica());
+ }
+
+ public static boolean isPreferredReplica(FetchResponseData.PartitionData partitionResponse) {
+ return partitionResponse.preferredReadReplica() != INVALID_PREFERRED_REPLICA_ID;
+ }
+
+ public static FetchResponseData.PartitionData partitionResponse(int partition, Errors error) {
+ return new FetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(error.code())
+ .setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK)
+ .setLastStableOffset(FetchResponse.INVALID_LAST_STABLE_OFFSET)
+ .setLogStartOffset(FetchResponse.INVALID_LOG_START_OFFSET)
+ .setAbortedTransactions(null)
+ .setRecords(MemoryRecords.EMPTY);
+ }
+
+ /**
+ * cast the BaseRecords of PartitionData to Records. This is used to eliminate duplicate code of type casting.
Review comment:
the data from KRPC always use `MemoryRecords` so it should never fail if the data is from KRPC. I will add more comments for this case.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#discussion_r585695430
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -761,79 +754,85 @@ class KafkaApis(val requestChannel: RequestChannel,
// For fetch requests from clients, check if down-conversion is disabled for the particular partition
if (!fetchRequest.isFromFollower && !logConfig.forall(_.messageDownConversionEnable)) {
trace(s"Conversion to message format ${downConvertMagic.get} is disabled for partition $tp. Sending unsupported version response to $clientId.")
- errorResponse(Errors.UNSUPPORTED_VERSION)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_VERSION)
} else {
try {
trace(s"Down converting records from partition $tp to message format version $magic for fetch request from $clientId")
// Because down-conversion is extremely memory intensive, we want to try and delay the down-conversion as much
// 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.
- val error = maybeDownConvertStorageError(partitionData.error)
- 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))
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(new LazyDownConversionRecords(tp, unconvertedRecords, magic, fetchContext.getFetchOffset(tp).get, time))
+ .setPreferredReadReplica(partitionData.preferredReadReplica())
} catch {
case e: UnsupportedCompressionTypeException =>
trace("Received unsupported compression type error during down-conversion", e)
- errorResponse(Errors.UNSUPPORTED_COMPRESSION_TYPE)
+ FetchResponse.partitionResponse(tp.partition, Errors.UNSUPPORTED_COMPRESSION_TYPE)
}
}
case None =>
- val error = maybeDownConvertStorageError(partitionData.error)
- new FetchResponse.PartitionData[BaseRecords](error,
- partitionData.highWatermark,
- partitionData.lastStableOffset,
- partitionData.logStartOffset,
- partitionData.preferredReadReplica,
- partitionData.abortedTransactions,
- partitionData.divergingEpoch,
- unconvertedRecords)
+ new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(Errors.forCode(partitionData.errorCode)).code)
+ .setHighWatermark(partitionData.highWatermark)
+ .setLastStableOffset(partitionData.lastStableOffset)
+ .setLogStartOffset(partitionData.logStartOffset)
+ .setAbortedTransactions(partitionData.abortedTransactions)
+ .setRecords(unconvertedRecords)
+ .setPreferredReadReplica(partitionData.preferredReadReplica)
+ .setDivergingEpoch(partitionData.divergingEpoch)
}
}
}
// the callback for process a fetch response, invoked before throttling
def processResponseCallback(responsePartitionData: Seq[(TopicPartition, FetchPartitionData)]): Unit = {
- val partitions = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[Records]]
+ val partitions = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
val reassigningPartitions = mutable.Set[TopicPartition]()
responsePartitionData.foreach { case (tp, data) =>
val abortedTransactions = data.abortedTransactions.map(_.asJava).orNull
val lastStableOffset = data.lastStableOffset.getOrElse(FetchResponse.INVALID_LAST_STABLE_OFFSET)
- if (data.isReassignmentFetch)
- reassigningPartitions.add(tp)
- val error = maybeDownConvertStorageError(data.error)
- partitions.put(tp, new FetchResponse.PartitionData(
- error,
- data.highWatermark,
- lastStableOffset,
- data.logStartOffset,
- data.preferredReadReplica.map(int2Integer).asJava,
- abortedTransactions,
- data.divergingEpoch.asJava,
- data.records))
+ if (data.isReassignmentFetch) reassigningPartitions.add(tp)
+ val partitionData = new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(maybeDownConvertStorageError(data.error).code)
+ .setHighWatermark(data.highWatermark)
+ .setLastStableOffset(lastStableOffset)
+ .setLogStartOffset(data.logStartOffset)
+ .setAbortedTransactions(abortedTransactions)
+ .setRecords(data.records)
+ .setPreferredReadReplica(data.preferredReadReplica.getOrElse(FetchResponse.INVALID_PREFERRED_REPLICA_ID))
+ data.divergingEpoch.foreach(partitionData.setDivergingEpoch)
+ partitions.put(tp, partitionData)
}
erroneous.foreach { case (tp, data) => partitions.put(tp, data) }
- var unconvertedFetchResponse: FetchResponse[Records] = null
+ var unconvertedFetchResponse: FetchResponse = null
- def createResponse(throttleTimeMs: Int): FetchResponse[BaseRecords] = {
+ def createResponse(throttleTimeMs: Int): FetchResponse = {
// Down-convert messages for each partition if required
- val convertedData = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData[BaseRecords]]
+ val convertedData = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]
unconvertedFetchResponse.responseData.forEach { (tp, unconvertedPartitionData) =>
- if (unconvertedPartitionData.error != Errors.NONE)
+ val error = Errors.forCode(unconvertedPartitionData.errorCode)
+ if (error != Errors.NONE)
debug(s"Fetch request with correlation id ${request.header.correlationId} from client $clientId " +
- s"on partition $tp failed due to ${unconvertedPartitionData.error.exceptionName}")
+ s"on partition $tp failed due to ${error.exceptionName}")
convertedData.put(tp, maybeConvertFetchedData(tp, unconvertedPartitionData))
}
// Prepare fetch response from converted data
- val response = new FetchResponse(unconvertedFetchResponse.error, convertedData, throttleTimeMs,
- unconvertedFetchResponse.sessionId)
+ val response = FetchResponse.of(unconvertedFetchResponse.error, throttleTimeMs, unconvertedFetchResponse.sessionId, convertedData)
// record the bytes out metrics only when the response is being sent
response.responseData.forEach { (tp, data) =>
- brokerTopicStats.updateBytesOut(tp.topic, fetchRequest.isFromFollower, reassigningPartitions.contains(tp), data.records.sizeInBytes)
+ brokerTopicStats.updateBytesOut(tp.topic, fetchRequest.isFromFollower,
+ reassigningPartitions.contains(tp), if (data.records == null) 0 else data.records.sizeInBytes)
Review comment:
> That would always be safe to call.
make sense. Will copy that
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on pull request #9758: MINOR: remove FetchResponse.AbortedTransaction and redundant construc…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9758:
URL: https://github.com/apache/kafka/pull/9758#issuecomment-757974418
fix conflicting files
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org