junrao commented on code in PR #18727:
URL: https://github.com/apache/kafka/pull/18727#discussion_r1938309781
##########
clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java:
##########
@@ -264,8 +269,11 @@ public boolean hasValidVersion() {
return oldestVersion() <= latestVersion();
}
- public Optional<ApiVersionsResponseData.ApiVersion> toApiVersion(boolean
enableUnstableLastVersion) {
- short oldestVersion = oldestVersion();
+ public Optional<ApiVersionsResponseData.ApiVersion> toApiVersion(boolean
enableUnstableLastVersion,
+
Optional<ApiMessageType.ListenerType> listenerType) {
Review Comment:
Since listenerType is optional, could we add a comment when the caller is
expected to pass in the right listenerType?
##########
core/src/test/scala/unit/kafka/network/ProcessorTest.scala:
##########
@@ -65,13 +65,31 @@ class ProcessorTest {
@Test
def testParseRequestHeaderWithUnsupportedApiVersion(): Unit = {
val requestHeader = RequestTestUtils.serializeRequestHeader(
- new RequestHeader(ApiKeys.PRODUCE, 0, "clientid", 0))
+ new RequestHeader(ApiKeys.FETCH, 0, "clientid", 0))
val apiVersionManager = new SimpleApiVersionManager(ListenerType.BROKER,
true,
() => new FinalizedFeatures(MetadataVersion.latestTesting(),
Collections.emptyMap[String, java.lang.Short], 0))
val e = assertThrows(classOf[UnsupportedVersionException],
(() => Processor.parseRequestHeader(apiVersionManager, requestHeader)):
Executable,
- "PRODUCE v0 should throw UnsupportedVersionException exception")
+ "FETCH v0 should throw UnsupportedVersionException exception")
assertTrue(e.toString.contains("unsupported version"));
}
+ /**
+ * We do something unusual with these versions of produce, and we want to
make sure we don't regress.
+ * See `ApiKeys.PRODUCE_API_VERSIONS_RESPONSE_MIN_VERSION` for details.
+ */
+ @Test
+ def testParseRequestHeaderForProduceV0ToV2(): Unit = {
+ for (version <- 0 to 2) {
+ val requestHeader = RequestTestUtils.serializeRequestHeader(
+ new RequestHeader(ApiKeys.PRODUCE, version.toShort, "clientid", 0))
+ val apiVersionManager = new SimpleApiVersionManager(ListenerType.BROKER,
true,
Review Comment:
This is an existing issue. Broker uses `DefaultApiVersionManager`, instead
of `SimpleApiVersionManager`. It would be useful to switch to
`DefaultApiVersionManager` here and in
`testParseRequestHeaderWithUnsupportedApiVersion()`.
--
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
