diff --git a/clients/src/main/resources/common/message/DescribeLogDirsResponse.json b/clients/src/main/resources/common/message/DescribeLogDirsResponse.json index 0171a16481f..c79e756aada 100644 --- a/clients/src/main/resources/common/message/DescribeLogDirsResponse.json +++ b/clients/src/main/resources/common/message/DescribeLogDirsResponse.json @@ -25,7 +25,8 @@ "fields": [ { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." }, - { "name": "ErrorCode", "type": "int16", "versions": "3+", "about": "The error code, or 0 if there was no error." }, + { "name": "ErrorCode", "type": "int16", "versions": "3+", + "ignorable": true, "about": "The error code, or 0 if there was no error." }, { "name": "Results", "type": "[]DescribeLogDirsResult", "versions": "0+", "about": "The log directories.", "fields": [ { "name": "ErrorCode", "type": "int16", "versions": "0+", diff --git a/core/src/test/scala/unit/kafka/network/RequestConvertToJsonTest.scala b/core/src/test/scala/unit/kafka/network/RequestConvertToJsonTest.scala index 09dceac1510..0ce8448a4f2 100644 --- a/core/src/test/scala/unit/kafka/network/RequestConvertToJsonTest.scala +++ b/core/src/test/scala/unit/kafka/network/RequestConvertToJsonTest.scala @@ -19,14 +19,13 @@ package kafka.network import java.net.InetAddress import java.nio.ByteBuffer - import com.fasterxml.jackson.databind.node.{BooleanNode, DoubleNode, JsonNodeFactory, LongNode, ObjectNode, TextNode} import kafka.network import kafka.network.RequestConvertToJson.requestHeaderNode import org.apache.kafka.common.memory.MemoryPool import org.apache.kafka.common.message._ import org.apache.kafka.common.network.{ClientInformation, ListenerName, NetworkSend} -import org.apache.kafka.common.protocol.{ApiKeys, MessageUtil} +import org.apache.kafka.common.protocol.{ApiKeys, Errors, MessageUtil} import org.apache.kafka.common.requests._ import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol} import org.junit.jupiter.api.Assertions.assertEquals @@ -61,6 +60,33 @@ class RequestConvertToJsonTest { assertEquals(ArrayBuffer.empty, unhandledKeys, "Unhandled request keys") } + @Test + def testAllApiVersionsResponseHandled(): Unit = { + + ApiKeys.values().foreach { key => { + val unhandledVersions = ArrayBuffer[java.lang.Short]() + key.allVersions().forEach { version => { + val message = key match { + // Specify top-level error handling for verifying compatibility across versions + case ApiKeys.DESCRIBE_LOG_DIRS => + ApiMessageType.fromApiKey(key.id).newResponse().asInstanceOf[DescribeLogDirsResponseData] + .setErrorCode(Errors.CLUSTER_AUTHORIZATION_FAILED.code()) + case _ => + ApiMessageType.fromApiKey(key.id).newResponse() + } + + val bytes = MessageUtil.toByteBuffer(message, version) + val response = AbstractResponse.parseResponse(key, bytes, version) + try { + RequestConvertToJson.response(response, version) + } catch { + case _ : IllegalStateException => unhandledVersions += version + }} + } + assertEquals(ArrayBuffer.empty, unhandledVersions, s"API: ${key.toString} - Unhandled request versions") + }} + } + @Test def testAllResponseTypesHandled(): Unit = { val unhandledKeys = ArrayBuffer[String]()