root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-12989 MockClient should respect the request matcher passed to prepareUnsupportedVersionResponse (#16849) 

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
This commit is contained in:
PoAn Yang 2024-08-19 23:19:01 +08:00 committed by GitHub
parent b63b20eeee
commit 2f0ae82d4a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 11 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
clients/src/test/java/org/apache/kafka/clients/MockClient.java
View File

@ -250,15 +250,16 @@ public class MockClient implements KafkaClient {
short version = nodeApiVersions.latestUsableVersion(request.apiKey(), builder.oldestAllowedVersion(),
builder.latestAllowedVersion());
AbstractRequest abstractRequest = request.requestBuilder().build(version);
if (!futureResp.requestMatcher.matches(abstractRequest))
throw new IllegalStateException("Request matcher did not match next-in-line request "
+ abstractRequest + " with prepared response " + futureResp.responseBody);
UnsupportedVersionException unsupportedVersionException = null;
if (futureResp.isUnsupportedRequest) {
unsupportedVersionException = new UnsupportedVersionException(
"Api " + request.apiKey() + " with version " + version);
} else {
AbstractRequest abstractRequest = request.requestBuilder().build(version);
if (!futureResp.requestMatcher.matches(abstractRequest))
throw new IllegalStateException("Request matcher did not match next-in-line request "
+ abstractRequest + " with prepared response " + futureResp.responseBody);
}
ClientResponse resp = new ClientResponse(request.makeHeader(version), request.callback(), request.destination(),

4
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java
View File

@ -876,7 +876,9 @@ public class TransactionManagerTest {
client.prepareUnsupportedVersionResponse(body -> {
FindCoordinatorRequest findCoordinatorRequest = (FindCoordinatorRequest) body;
assertEquals(CoordinatorType.forId(findCoordinatorRequest.data().keyType()), CoordinatorType.TRANSACTION);
assertEquals(findCoordinatorRequest.data().key(), transactionalId);
assertTrue(findCoordinatorRequest.data().key().isEmpty());
assertEquals(1, findCoordinatorRequest.data().coordinatorKeys().size());
assertTrue(findCoordinatorRequest.data().coordinatorKeys().contains(transactionalId));
return true;
});

3
core/src/test/scala/unit/kafka/server/ForwardingManagerTest.scala
View File

@ -171,7 +171,8 @@ class ForwardingManagerTest {
Mockito.when(controllerNodeProvider.getControllerInfo()).thenReturn(controllerInfo)
client.prepareUnsupportedVersionResponse(req => req.apiKey == requestHeader.apiKey)
val isEnvelopeRequest: RequestMatcher = request => request.isInstanceOf[EnvelopeRequest]
client.prepareUnsupportedVersionResponse(isEnvelopeRequest)
val response = new AtomicReference[AbstractResponse]()
forwardingManager.forwardRequest(request, res => res.foreach(response.set))