apalan60 commented on code in PR #19327:
URL: https://github.com/apache/kafka/pull/19327#discussion_r2023572891
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java:
##########
@@ -3900,86 +3653,6 @@ private Set<Node>
nodesRequested(List<NetworkClientDelegate.UnsentRequest> reque
.collect(Collectors.toSet());
}
- private FetchResponse fetchResponseWithTopLevelError(TopicIdPartition tp,
Errors error, int throttleTime) {
- Map<TopicIdPartition, FetchResponseData.PartitionData> partitions =
Collections.singletonMap(tp,
- new FetchResponseData.PartitionData()
- .setPartitionIndex(tp.topicPartition().partition())
- .setErrorCode(error.code())
-
.setHighWatermark(FetchResponse.INVALID_HIGH_WATERMARK));
- return FetchResponse.of(error, throttleTime, INVALID_SESSION_ID, new
LinkedHashMap<>(partitions));
- }
-
- private FetchResponse
fullFetchResponseWithAbortedTransactions(MemoryRecords records,
-
List<FetchResponseData.AbortedTransaction> abortedTransactions,
- Errors
error,
- long
lastStableOffset,
- long hw,
- int
throttleTime) {
- Map<TopicIdPartition, FetchResponseData.PartitionData> partitions =
Collections.singletonMap(tidp0,
- new FetchResponseData.PartitionData()
- .setPartitionIndex(tp0.partition())
- .setErrorCode(error.code())
- .setHighWatermark(hw)
- .setLastStableOffset(lastStableOffset)
- .setLogStartOffset(0)
- .setAbortedTransactions(abortedTransactions)
- .setRecords(records));
- return FetchResponse.of(Errors.NONE, throttleTime, INVALID_SESSION_ID,
new LinkedHashMap<>(partitions));
- }
-
- private FetchResponse fullFetchResponse(int sessionId, TopicIdPartition
tp, MemoryRecords records, Errors error, long hw, int throttleTime) {
- return fullFetchResponse(sessionId, tp, records, error, hw,
FetchResponse.INVALID_LAST_STABLE_OFFSET, throttleTime);
- }
-
- private FetchResponse fullFetchResponse(TopicIdPartition tp, MemoryRecords
records, Errors error, long hw, int throttleTime) {
- return fullFetchResponse(tp, records, error, hw,
FetchResponse.INVALID_LAST_STABLE_OFFSET, throttleTime);
- }
-
- private FetchResponse fullFetchResponse(TopicIdPartition tp, MemoryRecords
records, Errors error, long hw,
- long lastStableOffset, int
throttleTime) {
- return fullFetchResponse(INVALID_SESSION_ID, tp, records, error, hw,
lastStableOffset, throttleTime);
- }
-
- private FetchResponse fullFetchResponse(int sessionId, TopicIdPartition
tp, MemoryRecords records, Errors error, long hw,
- long lastStableOffset, int
throttleTime) {
- Map<TopicIdPartition, FetchResponseData.PartitionData> partitions =
Collections.singletonMap(tp,
- new FetchResponseData.PartitionData()
- .setPartitionIndex(tp.topicPartition().partition())
- .setErrorCode(error.code())
- .setHighWatermark(hw)
- .setLastStableOffset(lastStableOffset)
- .setLogStartOffset(0)
- .setRecords(records));
- return FetchResponse.of(Errors.NONE, throttleTime, sessionId, new
LinkedHashMap<>(partitions));
Review Comment:
@FrankYang0529
Thanks for the review.
I really appreciate your detailed explanation and pointing out my mistake.
I’ll try out your suggested solution and update the PR accordingly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
