diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java index 428e4a81b6835..31392e40a1474 100644 --- a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java +++ b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java @@ -98,7 +98,7 @@ public enum ApiKeys { END_QUORUM_EPOCH(ApiMessageType.END_QUORUM_EPOCH, true, RecordBatch.MAGIC_VALUE_V0, false), DESCRIBE_QUORUM(ApiMessageType.DESCRIBE_QUORUM, true, RecordBatch.MAGIC_VALUE_V0, true), ALTER_ISR(ApiMessageType.ALTER_ISR, true), - UPDATE_FEATURES(ApiMessageType.UPDATE_FEATURES, false, true), + UPDATE_FEATURES(ApiMessageType.UPDATE_FEATURES), ENVELOPE(ApiMessageType.ENVELOPE, true, RecordBatch.MAGIC_VALUE_V0, false), FETCH_SNAPSHOT(ApiMessageType.FETCH_SNAPSHOT, false, RecordBatch.MAGIC_VALUE_V0, false), DESCRIBE_CLUSTER(ApiMessageType.DESCRIBE_CLUSTER), @@ -267,6 +267,14 @@ public static EnumSet zkBrokerApis() { return apisForListener(ApiMessageType.ListenerType.ZK_BROKER); } + public static EnumSet controllerApis() { + return apisForListener(ApiMessageType.ListenerType.CONTROLLER); + } + + public static EnumSet brokerApis() { + return apisForListener(ApiMessageType.ListenerType.BROKER); + } + public static EnumSet apisForListener(ApiMessageType.ListenerType listener) { return APIS_BY_LISTENER.get(listener); } diff --git a/core/src/main/scala/kafka/server/AlterIsrManager.scala b/core/src/main/scala/kafka/server/AlterIsrManager.scala index b8507d086d8d2..0d94ceb98d803 100644 --- a/core/src/main/scala/kafka/server/AlterIsrManager.scala +++ b/core/src/main/scala/kafka/server/AlterIsrManager.scala @@ -27,7 +27,7 @@ import kafka.zk.KafkaZkClient import org.apache.kafka.clients.ClientResponse import org.apache.kafka.common.TopicPartition import org.apache.kafka.common.errors.OperationNotAttemptedException -import org.apache.kafka.common.message.{AlterIsrRequestData, AlterIsrResponseData} +import org.apache.kafka.common.message.{AlterIsrRequestData, AlterIsrResponseData, ApiMessageType} import org.apache.kafka.common.metrics.Metrics import org.apache.kafka.common.protocol.Errors import org.apache.kafka.common.requests.{AlterIsrRequest, AlterIsrResponse} @@ -86,7 +86,8 @@ object AlterIsrManager { config = config, channelName = "alterIsr", threadNamePrefix = threadNamePrefix, - retryTimeoutMs = Long.MaxValue + retryTimeoutMs = Long.MaxValue, + ApiMessageType.ListenerType.ZK_BROKER ) new DefaultAlterIsrManager( controllerChannelManager = channelManager, diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala b/core/src/main/scala/kafka/server/BrokerServer.scala index b3b8e8382f2af..01c148ace9be2 100644 --- a/core/src/main/scala/kafka/server/BrokerServer.scala +++ b/core/src/main/scala/kafka/server/BrokerServer.scala @@ -218,7 +218,8 @@ class BrokerServer( config, channelName = "forwarding", threadNamePrefix, - retryTimeoutMs = 60000 + retryTimeoutMs = 60000, + ListenerType.BROKER ) clientToControllerChannelManager.start() forwardingManager = new ForwardingManagerImpl(clientToControllerChannelManager) @@ -246,7 +247,8 @@ class BrokerServer( config, channelName = "alterIsr", threadNamePrefix, - retryTimeoutMs = Long.MaxValue + retryTimeoutMs = Long.MaxValue, + ListenerType.BROKER ) alterIsrManager = new DefaultAlterIsrManager( controllerChannelManager = alterIsrChannelManager, @@ -333,7 +335,7 @@ class BrokerServer( } lifecycleManager.start(() => metadataListener.highestMetadataOffset, BrokerToControllerChannelManager(controllerNodeProvider, time, metrics, config, - "heartbeat", threadNamePrefix, config.brokerSessionTimeoutMs.toLong), + "heartbeat", threadNamePrefix, config.brokerSessionTimeoutMs.toLong, ListenerType.BROKER), metaProps.clusterId, networkListeners, supportedFeatures) // Register a listener with the Raft layer to receive metadata event notifications diff --git a/core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala b/core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala index 132a2cc98b06a..b345644da24d0 100644 --- a/core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala +++ b/core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala @@ -19,16 +19,16 @@ package kafka.server import java.util.concurrent.LinkedBlockingDeque import java.util.concurrent.atomic.AtomicReference - import kafka.common.{InterBrokerSendThread, RequestAndCompletionHandler} import kafka.raft.RaftManager import kafka.utils.Logging import org.apache.kafka.clients._ import org.apache.kafka.common.Node +import org.apache.kafka.common.message.ApiMessageType import org.apache.kafka.common.metrics.Metrics import org.apache.kafka.common.network._ -import org.apache.kafka.common.protocol.Errors -import org.apache.kafka.common.requests.AbstractRequest +import org.apache.kafka.common.protocol.{ApiKeys, Errors} +import org.apache.kafka.common.requests.{AbstractRequest, ApiVersionsResponse} import org.apache.kafka.common.security.JaasContext import org.apache.kafka.common.security.auth.SecurityProtocol import org.apache.kafka.common.utils.{LogContext, Time} @@ -119,7 +119,8 @@ object BrokerToControllerChannelManager { config: KafkaConfig, channelName: String, threadNamePrefix: Option[String], - retryTimeoutMs: Long + retryTimeoutMs: Long, + controllerBrokerType: ApiMessageType.ListenerType ): BrokerToControllerChannelManager = { new BrokerToControllerChannelManagerImpl( controllerNodeProvider, @@ -128,7 +129,8 @@ object BrokerToControllerChannelManager { config, channelName, threadNamePrefix, - retryTimeoutMs + retryTimeoutMs, + controllerBrokerType ) } } @@ -159,12 +161,13 @@ class BrokerToControllerChannelManagerImpl( config: KafkaConfig, channelName: String, threadNamePrefix: Option[String], - retryTimeoutMs: Long + retryTimeoutMs: Long, + controllerBrokerType: ApiMessageType.ListenerType ) extends BrokerToControllerChannelManager with Logging { private val logContext = new LogContext(s"[BrokerToControllerChannelManager broker=${config.brokerId} name=$channelName] ") private val manualMetadataUpdater = new ManualMetadataUpdater() private val apiVersions = new ApiVersions() - private val currentNodeApiVersions = NodeApiVersions.create() + private val currentNodeControllerApiVersions = new NodeApiVersions(ApiKeys.apisForListener(controllerBrokerType).asScala.map(ApiVersionsResponse.toApiVersion).asJava) private val requestThread = newRequestThread def start(): Unit = { @@ -253,7 +256,7 @@ class BrokerToControllerChannelManagerImpl( def controllerApiVersions(): Option[NodeApiVersions] = requestThread.activeControllerAddress().flatMap( activeController => if (activeController.id() == config.brokerId) - Some(currentNodeApiVersions) + Some(currentNodeControllerApiVersions) else Option(apiVersions.get(activeController.idString())) ) diff --git a/core/src/main/scala/kafka/server/ControllerApis.scala b/core/src/main/scala/kafka/server/ControllerApis.scala index ed9b55a1a7c39..f4a02106dfe00 100644 --- a/core/src/main/scala/kafka/server/ControllerApis.scala +++ b/core/src/main/scala/kafka/server/ControllerApis.scala @@ -183,7 +183,7 @@ class ControllerApis(val requestChannel: RequestChannel, throw new TopicDeletionDisabledException() } } - val deadlineNs = time.nanoseconds() + NANOSECONDS.convert(request.timeoutMs, MILLISECONDS); + val deadlineNs = time.nanoseconds() + NANOSECONDS.convert(request.timeoutMs, MILLISECONDS) // The first step is to load up the names and IDs that have been provided by the // request. This is a bit messy because we support multiple ways of referring to // topics (both by name and by id) and because we need to check for duplicates or @@ -704,7 +704,7 @@ class ControllerApis(val requestChannel: RequestChannel, hasClusterAuth: Boolean, getCreatableTopics: Iterable[String] => Set[String]) : CompletableFuture[util.List[CreatePartitionsTopicResult]] = { - val deadlineNs = time.nanoseconds() + NANOSECONDS.convert(request.timeoutMs, MILLISECONDS); + val deadlineNs = time.nanoseconds() + NANOSECONDS.convert(request.timeoutMs, MILLISECONDS) val responses = new util.ArrayList[CreatePartitionsTopicResult]() val duplicateTopicNames = new util.HashSet[String]() val topicNames = new util.HashSet[String]() diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index e20a36a0e3ebd..2968b3d465202 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/core/src/main/scala/kafka/server/KafkaApis.scala @@ -219,7 +219,7 @@ class KafkaApis(val requestChannel: RequestChannel, case ApiKeys.DESCRIBE_USER_SCRAM_CREDENTIALS => handleDescribeUserScramCredentialsRequest(request) case ApiKeys.ALTER_USER_SCRAM_CREDENTIALS => maybeForwardToController(request, handleAlterUserScramCredentialsRequest) case ApiKeys.ALTER_ISR => handleAlterIsrRequest(request) - case ApiKeys.UPDATE_FEATURES => maybeForwardToController(request, handleUpdateFeatures) + case ApiKeys.UPDATE_FEATURES => handleUpdateFeatures(request) case ApiKeys.ENVELOPE => handleEnvelope(request, requestLocal) case ApiKeys.DESCRIBE_CLUSTER => handleDescribeCluster(request) case ApiKeys.DESCRIBE_PRODUCERS => handleDescribeProducersRequest(request) @@ -3244,7 +3244,7 @@ class KafkaApis(val requestChannel: RequestChannel, } def handleUpdateFeatures(request: RequestChannel.Request): Unit = { - val zkSupport = metadataSupport.requireZkOrThrow(KafkaApis.shouldAlwaysForward(request)) + val zkSupport = metadataSupport.requireZkOrThrow(KafkaApis.notYetSupported(request)) val updateFeaturesRequest = request.body[UpdateFeaturesRequest] def sendResponseCallback(errors: Either[ApiError, Map[String, ApiError]]): Unit = { diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala index 19ede1b41946c..9ea1f7a8e711b 100755 --- a/core/src/main/scala/kafka/server/KafkaServer.scala +++ b/core/src/main/scala/kafka/server/KafkaServer.scala @@ -269,7 +269,9 @@ class KafkaServer( config = config, channelName = "forwarding", threadNamePrefix = threadNamePrefix, - retryTimeoutMs = config.requestTimeoutMs.longValue) + retryTimeoutMs = config.requestTimeoutMs.longValue, + ListenerType.ZK_BROKER + ) clientToControllerChannelManager.start() /* start forwarding manager */ diff --git a/core/src/test/scala/unit/kafka/server/AbstractApiVersionsRequestTest.scala b/core/src/test/scala/unit/kafka/server/AbstractApiVersionsRequestTest.scala index 1e6c05d9a487a..b0af7871d3798 100644 --- a/core/src/test/scala/unit/kafka/server/AbstractApiVersionsRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/AbstractApiVersionsRequestTest.scala @@ -16,13 +16,15 @@ */ package kafka.server -import java.util.Properties - import kafka.test.ClusterInstance +import kafka.test.ClusterInstance.ClusterType +import org.apache.kafka.clients.NodeApiVersions import org.apache.kafka.common.message.ApiMessageType.ListenerType import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersion +import org.apache.kafka.common.message.{ApiMessageType, ApiVersionsResponseData} import org.apache.kafka.common.network.ListenerName import org.apache.kafka.common.protocol.ApiKeys +import org.apache.kafka.common.record.RecordVersion import org.apache.kafka.common.requests.{ApiVersionsRequest, ApiVersionsResponse, RequestUtils} import org.apache.kafka.common.utils.Utils import org.junit.jupiter.api.Assertions._ @@ -39,15 +41,6 @@ abstract class AbstractApiVersionsRequestTest(cluster: ClusterInstance) { def controlPlaneListenerName = new ListenerName("CONTROLLER") - // Configure control plane listener to make sure we have separate listeners for testing. - def brokerPropertyOverrides(properties: Properties): Unit = { - val securityProtocol = cluster.config().securityProtocol() - properties.setProperty(KafkaConfig.ControlPlaneListenerNameProp, controlPlaneListenerName.value()) - properties.setProperty(KafkaConfig.ListenerSecurityProtocolMapProp, s"${controlPlaneListenerName.value()}:$securityProtocol,$securityProtocol:$securityProtocol") - properties.setProperty("listeners", s"$securityProtocol://localhost:0,${controlPlaneListenerName.value()}://localhost:0") - properties.setProperty(KafkaConfig.AdvertisedListenersProp, s"$securityProtocol://localhost:0,${controlPlaneListenerName.value()}://localhost:0") - } - def sendUnsupportedApiVersionRequest(request: ApiVersionsRequest): ApiVersionsResponse = { val overrideHeader = IntegrationTestUtils.nextRequestHeader(ApiKeys.API_VERSIONS, Short.MaxValue) val socket = IntegrationTestUtils.connect(cluster.brokerSocketServers().asScala.head, cluster.clientListener()) @@ -60,14 +53,34 @@ abstract class AbstractApiVersionsRequestTest(cluster: ClusterInstance) { } def validateApiVersionsResponse(apiVersionsResponse: ApiVersionsResponse): Unit = { - val expectedApis = ApiKeys.zkBrokerApis() + val expectedApis = if (cluster.clusterType() == ClusterType.ZK) + ApiKeys.zkBrokerApis() + else { + val apis = ApiVersionsResponse.intersectForwardableApis( + ApiMessageType.ListenerType.BROKER, + RecordVersion.current, + new NodeApiVersions(ApiKeys.controllerApis().asScala.map(ApiVersionsResponse.toApiVersion).asJava).allSupportedApiVersions() + ) + apis.add( + new ApiVersionsResponseData.ApiVersion() + .setApiKey(ApiKeys.ENVELOPE.id) + .setMinVersion(ApiKeys.ENVELOPE.oldestVersion) + .setMaxVersion(ApiKeys.ENVELOPE.latestVersion) + ) + apis + } + assertEquals(expectedApis.size(), apiVersionsResponse.data.apiKeys().size(), "API keys in ApiVersionsResponse must match API keys supported by broker.") - val defaultApiVersionsResponse = ApiVersionsResponse.defaultApiVersionsResponse(ListenerType.ZK_BROKER) + val defaultApiVersionsResponse = if (cluster.clusterType() == ClusterType.ZK) + ApiVersionsResponse.defaultApiVersionsResponse(ListenerType.ZK_BROKER) + else + ApiVersionsResponse.createApiVersionsResponse(0, expectedApis.asInstanceOf[ApiVersionsResponseData.ApiVersionCollection]) + for (expectedApiVersion: ApiVersion <- defaultApiVersionsResponse.data.apiKeys().asScala) { val actualApiVersion = apiVersionsResponse.apiVersion(expectedApiVersion.apiKey) - assertNotNull(actualApiVersion, s"API key ${actualApiVersion.apiKey} is supported by broker, but not received in ApiVersionsResponse.") + assertNotNull(actualApiVersion, s"API key ${expectedApiVersion.apiKey()} is supported by broker, but not received in ApiVersionsResponse.") assertEquals(expectedApiVersion.apiKey, actualApiVersion.apiKey, "API key must be supported by the broker.") assertEquals(expectedApiVersion.minVersion, actualApiVersion.minVersion, s"Received unexpected min version for API key ${actualApiVersion.apiKey}.") assertEquals(expectedApiVersion.maxVersion, actualApiVersion.maxVersion, s"Received unexpected max version for API key ${actualApiVersion.apiKey}.") diff --git a/core/src/test/scala/unit/kafka/server/ApiVersionsRequestTest.scala b/core/src/test/scala/unit/kafka/server/ApiVersionsRequestTest.scala index 34ee74a0e8066..1fa60be7bc8c6 100644 --- a/core/src/test/scala/unit/kafka/server/ApiVersionsRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/ApiVersionsRequestTest.scala @@ -17,25 +17,20 @@ package kafka.server -import kafka.test.{ClusterConfig, ClusterInstance} +import kafka.test.ClusterInstance +import kafka.test.annotation.{ClusterTest, ClusterTestDefaults, Type} +import kafka.test.junit.ClusterTestExtensions import org.apache.kafka.common.message.ApiVersionsRequestData import org.apache.kafka.common.protocol.{ApiKeys, Errors} import org.apache.kafka.common.requests.ApiVersionsRequest -import kafka.test.annotation.ClusterTest -import kafka.test.junit.ClusterTestExtensions import org.junit.jupiter.api.Assertions._ -import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.extension.ExtendWith @ExtendWith(value = Array(classOf[ClusterTestExtensions])) +@ClusterTestDefaults(clusterType = Type.BOTH, brokers = 1) class ApiVersionsRequestTest(cluster: ClusterInstance) extends AbstractApiVersionsRequestTest(cluster) { - @BeforeEach - def setup(config: ClusterConfig): Unit = { - super.brokerPropertyOverrides(config.serverProperties()) - } - @ClusterTest def testApiVersionsRequest(): Unit = { val request = new ApiVersionsRequest.Builder().build() @@ -46,7 +41,7 @@ class ApiVersionsRequestTest(cluster: ClusterInstance) extends AbstractApiVersio @ClusterTest def testApiVersionsRequestThroughControlPlaneListener(): Unit = { val request = new ApiVersionsRequest.Builder().build() - val apiVersionsResponse = sendApiVersionsRequest(request, super.controlPlaneListenerName) + val apiVersionsResponse = sendApiVersionsRequest(request, cluster.clientListener) validateApiVersionsResponse(apiVersionsResponse) } @@ -72,7 +67,7 @@ class ApiVersionsRequestTest(cluster: ClusterInstance) extends AbstractApiVersio @ClusterTest def testApiVersionsRequestValidationV0ThroughControlPlaneListener(): Unit = { val apiVersionsRequest = new ApiVersionsRequest.Builder().build(0.asInstanceOf[Short]) - val apiVersionsResponse = sendApiVersionsRequest(apiVersionsRequest, super.controlPlaneListenerName) + val apiVersionsResponse = sendApiVersionsRequest(apiVersionsRequest, cluster.clientListener) validateApiVersionsResponse(apiVersionsResponse) } diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala index fb77d5bed0ae7..f26dd763dfe1a 100644 --- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala +++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala @@ -893,12 +893,6 @@ class KafkaApisTest { testForwardableApi(ApiKeys.DELETE_TOPICS, requestBuilder) } - @Test - def testUpdateFeaturesWithForwarding(): Unit = { - val requestBuilder = new UpdateFeaturesRequest.Builder(new UpdateFeaturesRequestData()) - testForwardableApi(ApiKeys.UPDATE_FEATURES, requestBuilder) - } - @Test def testAlterScramWithForwarding(): Unit = { val requestBuilder = new AlterUserScramCredentialsRequest.Builder(new AlterUserScramCredentialsRequestData()) @@ -4137,12 +4131,6 @@ class KafkaApisTest { verifyShouldAlwaysForwardErrorMessage(createKafkaApis(raftSupport = true).handleAlterUserScramCredentialsRequest) } - @Test - def testRaftShouldAlwaysForwardUpdateFeatures(): Unit = { - metadataCache = MetadataCache.kRaftMetadataCache(brokerId) - verifyShouldAlwaysForwardErrorMessage(createKafkaApis(raftSupport = true).handleUpdateFeatures) - } - @Test def testRaftShouldAlwaysForwardElectLeaders(): Unit = { metadataCache = MetadataCache.kRaftMetadataCache(brokerId) diff --git a/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala b/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala index 26609480443db..da370cfb0c60c 100644 --- a/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala @@ -49,7 +49,6 @@ class SaslApiVersionsRequestTest(cluster: ClusterInstance) extends AbstractApiVe sasl.startSasl(sasl.jaasSections(kafkaServerSaslMechanisms, Some(kafkaClientSaslMechanism), KafkaSasl, JaasTestUtils.KafkaServerContextName)) config.saslServerProperties().putAll(sasl.kafkaServerSaslProperties(kafkaServerSaslMechanisms, kafkaClientSaslMechanism)) config.saslClientProperties().putAll(sasl.kafkaClientSaslProperties(kafkaClientSaslMechanism)) - super.brokerPropertyOverrides(config.serverProperties()) } @ClusterTest(securityProtocol = SecurityProtocol.SASL_PLAINTEXT, clusterType = Type.ZK) diff --git a/raft/src/main/java/org/apache/kafka/raft/RaftConfig.java b/raft/src/main/java/org/apache/kafka/raft/RaftConfig.java index 0833df0bb2328..a687c4e8a4cff 100644 --- a/raft/src/main/java/org/apache/kafka/raft/RaftConfig.java +++ b/raft/src/main/java/org/apache/kafka/raft/RaftConfig.java @@ -264,7 +264,7 @@ public void ensureValid(String name, Object value) { } @SuppressWarnings("unchecked") - List voterStrings = (List) value; + List voterStrings = (List) value; // Attempt to parse the connect strings parseVoterConnections(voterStrings);