KAFKA-19844: ShareFetch related changes to support RENEW. #20826
Conversation
| erroneousTopicIdPartitions.add(tp) | ||
| isErroneous = true | ||
| } else if (batch.acknowledgeTypes.stream().anyMatch(ackType => ackType < 0 || ackType > 3)) { | ||
| } else if (batch.acknowledgeTypes.stream().anyMatch(ackType => ackType < 0 || ackType > 4)) { |
There was a problem hiding this comment.
Personally, I would add a pair of parameters, one which indicates whether the request is permitted to use RENEW (so version >= 2), and another to indicate whether the request actually has isRenewAck set. If the request version is >= 2, then the max ack type is 4, else 3. If isRenewAck is not set, the ackType must not be 4.
| val fetchResult: CompletableFuture[Map[TopicIdPartition, ShareFetchResponseData.PartitionData]] = | ||
| if (shareFetchRequest.version() >= 2 && shareFetchRequest.data.isRenewAck) | ||
| CompletableFuture.completedFuture(mutable.Map.empty[TopicIdPartition, ShareFetchResponseData.PartitionData]) | ||
| else | ||
| handleFetchFromShareFetchRequest(request, shareSessionEpoch, erroneousAndValidPartitionData, sharePartitionManager, authorizedTopics) |
There was a problem hiding this comment.
Can we please write comments for the changes.
| when(sharePartitionManager.fetchMessages(any(), any(), any(), anyInt(), anyInt(), anyInt(), any())).thenReturn( | ||
| CompletableFuture.completedFuture(util.Map.of[TopicIdPartition, ShareFetchResponseData.PartitionData]( | ||
| new TopicIdPartition(topicId, new TopicPartition(topicName, partitionIndex)), | ||
| new ShareFetchResponseData.PartitionData() | ||
| .setErrorCode(Errors.NONE.code) | ||
| .setAcknowledgeErrorCode(Errors.NONE.code) | ||
| .setRecords(records1) | ||
| .setAcquiredRecords(new util.ArrayList(util.List.of( | ||
| new ShareFetchResponseData.AcquiredRecords() | ||
| .setFirstOffset(0) | ||
| .setLastOffset(9) | ||
| .setDeliveryCount(1) | ||
| ))) | ||
| )) | ||
| ).thenReturn( |
There was a problem hiding this comment.
Why do we need any mocks for sharePartitionManager.fetchMessage if we just want to test that no fetch happens when renew ack is present? We can simply check the mocked sharePartitionManager that no calls should happen for fetchMessages, correct?
There was a problem hiding this comment.
Yes, I was planning to add a not-called assertion but it isn't needed. There should be one mock only. The test simulates one sharefetch to get records and then call renew on one of those records.
|
@AndrewJSchofield @apoorvmittal10 Thanks for the review, incorporated comments. |
There was a problem hiding this comment.
Overall source code looks good. I would recommend adding some tests for share acknowledge RPC handling along with RENEW ack. For some tests, we should use multiple Acknowledgement types in acknowledgement batches just to make sure everything is working right.
| new ShareSessionKey(groupId, memberId), cachedSharePartitions, 2)) | ||
| ) | ||
|
|
||
| when(clientQuotaManager.maybeRecordAndGetThrottleTimeMs( |
There was a problem hiding this comment.
We can get rid of this mock. We don't need it.
| } | ||
|
|
||
| @Test | ||
| def testHandleShareFetchRequestSuccessWithRenewAcknowledgements(): Unit = { |
There was a problem hiding this comment.
I think we should enhance this test case by using both RENEW acks and the previous present acknowledgements just to make sure things are working fine
There was a problem hiding this comment.
It will not make much difference as we are mocking the meat of the method - sharePartitionManager.fetchMessages. Will add nevertheless.
| val response = verifyNoThrottling[ShareFetchResponse](request) | ||
| val responseData = response.data() | ||
|
|
||
| assertEquals(Errors.INVALID_REQUEST.code, responseData.errorCode) |
There was a problem hiding this comment.
In this case we should also add asserts over the error message, since you have custom messages for different conditions
There was a problem hiding this comment.
@adixitconfluent
I was planning to do this initially but
There is a shortcoming in Errors class where the message contained in e is swallowed (Errors.forException(e)) in ShareFetchRequest.getErrorResponse. Error responses from other requests have the same issue as well where only the error code is set.
ErrorCode is way to go. In general it might not be best practise to assert on specific messages as they might change often. Error codes should be verified.
|
|
||
| val groupId = "group" | ||
|
|
||
| when(clientQuotaManager.maybeRecordAndGetThrottleTimeMs( |
There was a problem hiding this comment.
We don't need this mock. We should get rid of it.
4 participants
sub-routine is skipped in the KafkaApis handler for the request.
been added along with validations on fields maxBytes, minBytes,
maxRecords and maxWaitMs which should be set to 0 for version >= 2 and
isRenewAck set to true.