FrankYang0529 commented on code in PR #19327:
URL: https://github.com/apache/kafka/pull/19327#discussion_r2022112558
##########
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:
The `FetchResponse.of` is not used in production code, but we still need to
use another function to replace it in test code. Probably, you can modify this
kind of function like following and add related `client.prepareResponse` back
in the test code.
```java
private FetchResponse fullFetchResponse(int sessionId, TopicIdPartition
tp, MemoryRecords records, Errors error, long hw,
long lastStableOffset, int
throttleTime) {
return new FetchResponse(new FetchResponseData()
.setThrottleTimeMs(throttleTime)
.setErrorCode(Errors.NONE.code())
.setSessionId(sessionId)
.setResponses(List.of(
new FetchResponseData.FetchableTopicResponse()
.setTopic(tp.topic())
.setTopicId(tp.topicId())
.setPartitions(List.of(
new FetchResponseData.PartitionData()
.setPartitionIndex(tp.topicPartition().partition())
.setErrorCode(error.code())
.setHighWatermark(hw)
.setLastStableOffset(lastStableOffset)
.setLogStartOffset(0)
.setRecords(records))
))));
}
```
--
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
