diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java index 7b896ffe0177c..d1f6e68c6536a 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java @@ -1550,7 +1550,7 @@ public void handleResponse(AbstractResponse response) { } else if (findCoordinatorResponse.error() == Errors.GROUP_AUTHORIZATION_FAILED) { abortableError(GroupAuthorizationException.forGroupId(builder.data().key())); } else { - fatalError(new KafkaException(String.format("Could not find a coordinator with type %s with key %s due to" + + fatalError(new KafkaException(String.format("Could not find a coordinator with type %s with key %s due to " + "unexpected error: %s", coordinatorType, builder.data().key(), findCoordinatorResponse.data().errorMessage()))); } diff --git a/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java index 9a1032b8c0955..a003c2d3f7a59 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java @@ -65,6 +65,16 @@ public CreateTopicsRequest build(short version) { public String toString() { return data.toString(); } + + @Override + public boolean equals(Object other) { + return other instanceof Builder && this.data.equals(((Builder) other).data); + } + + @Override + public int hashCode() { + return data.hashCode(); + } } private final CreateTopicsRequestData data; diff --git a/clients/src/main/java/org/apache/kafka/common/requests/FindCoordinatorRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/FindCoordinatorRequest.java index dc512a554e246..f4491ad1ee5b1 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/FindCoordinatorRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/FindCoordinatorRequest.java @@ -17,6 +17,7 @@ package org.apache.kafka.common.requests; import org.apache.kafka.common.Node; +import org.apache.kafka.common.errors.InvalidRequestException; import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.message.FindCoordinatorRequestData; import org.apache.kafka.common.message.FindCoordinatorResponseData; @@ -102,7 +103,7 @@ public static CoordinatorType forId(byte id) { case 1: return TRANSACTION; default: - throw new IllegalArgumentException("Unknown coordinator type received: " + id); + throw new InvalidRequestException("Unknown coordinator type received: " + id); } } } diff --git a/clients/src/test/java/org/apache/kafka/common/requests/FindCoordinatorRequestTest.java b/clients/src/test/java/org/apache/kafka/common/requests/FindCoordinatorRequestTest.java new file mode 100644 index 0000000000000..9393358074120 --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/common/requests/FindCoordinatorRequestTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.common.requests; + +import org.apache.kafka.common.errors.InvalidRequestException; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertThrows; + +class FindCoordinatorRequestTest { + + @Test + public void getInvalidCoordinatorTypeId() { + assertThrows(InvalidRequestException.class, + () -> FindCoordinatorRequest.CoordinatorType.forId((byte) 10)); + } +} diff --git a/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala b/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala new file mode 100644 index 0000000000000..ec7f2df3e6cdc --- /dev/null +++ b/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala @@ -0,0 +1,321 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package kafka.server + +import java.util.{Collections, Properties} +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicReference + +import kafka.controller.KafkaController +import kafka.coordinator.group.GroupCoordinator +import kafka.coordinator.transaction.TransactionCoordinator +import kafka.server.metadata.MetadataBroker +import kafka.utils.Logging +import org.apache.kafka.clients.ClientResponse +import org.apache.kafka.common.errors.InvalidTopicException +import org.apache.kafka.common.internals.Topic +import org.apache.kafka.common.internals.Topic.{GROUP_METADATA_TOPIC_NAME, TRANSACTION_STATE_TOPIC_NAME} +import org.apache.kafka.common.message.CreateTopicsRequestData +import org.apache.kafka.common.message.CreateTopicsRequestData.{CreatableTopic, CreateableTopicConfig, CreateableTopicConfigCollection} +import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseTopic +import org.apache.kafka.common.metrics.Metrics +import org.apache.kafka.common.protocol.Errors +import org.apache.kafka.common.requests.{ApiError, CreateTopicsRequest} +import org.apache.kafka.common.utils.Time + +import scala.collection.{Map, Seq, Set, mutable} +import scala.jdk.CollectionConverters._ + +trait AutoTopicCreationManager { + + def createTopics( + topicNames: Set[String], + controllerMutationQuota: ControllerMutationQuota + ): Seq[MetadataResponseTopic] + + def start(): Unit + + def shutdown(): Unit +} + +object AutoTopicCreationManager { + + def apply( + config: KafkaConfig, + metadataCache: MetadataCache, + time: Time, + metrics: Metrics, + threadNamePrefix: Option[String], + adminManager: ZkAdminManager, + controller: KafkaController, + groupCoordinator: GroupCoordinator, + txnCoordinator: TransactionCoordinator, + enableForwarding: Boolean + ): AutoTopicCreationManager = { + + val channelManager = + if (enableForwarding) + Some(new BrokerToControllerChannelManagerImpl( + controllerNodeProvider = MetadataCacheControllerNodeProvider( + config, metadataCache), + time = time, + metrics = metrics, + config = config, + channelName = "autoTopicCreationChannel", + threadNamePrefix = threadNamePrefix, + retryTimeoutMs = config.requestTimeoutMs.longValue + )) + else + None + new DefaultAutoTopicCreationManager(config, metadataCache, channelManager, adminManager, + controller, groupCoordinator, txnCoordinator) + } +} + +class DefaultAutoTopicCreationManager( + config: KafkaConfig, + metadataCache: MetadataCache, + channelManager: Option[BrokerToControllerChannelManager], + adminManager: ZkAdminManager, + controller: KafkaController, + groupCoordinator: GroupCoordinator, + txnCoordinator: TransactionCoordinator +) extends AutoTopicCreationManager with Logging { + + private val inflightTopics = Collections.newSetFromMap(new ConcurrentHashMap[String, java.lang.Boolean]()) + + override def start(): Unit = { + channelManager.foreach(_.start()) + } + + override def shutdown(): Unit = { + channelManager.foreach(_.shutdown()) + inflightTopics.clear() + } + + override def createTopics( + topics: Set[String], + controllerMutationQuota: ControllerMutationQuota + ): Seq[MetadataResponseTopic] = { + val (creatableTopics, uncreatableTopicResponses) = filterCreatableTopics(topics) + + val creatableTopicResponses = if (creatableTopics.isEmpty) { + Seq.empty + } else if (!controller.isActive && channelManager.isDefined) { + sendCreateTopicRequest(creatableTopics) + } else { + createTopicsInZk(creatableTopics, controllerMutationQuota) + } + + uncreatableTopicResponses ++ creatableTopicResponses + } + + private def createTopicsInZk( + creatableTopics: Map[String, CreatableTopic], + controllerMutationQuota: ControllerMutationQuota + ): Seq[MetadataResponseTopic] = { + val topicErrors = new AtomicReference[Map[String, ApiError]]() + try { + // Note that we use timeout = 0 since we do not need to wait for metadata propagation + // and we want to get the response error immediately. + adminManager.createTopics( + timeout = 0, + validateOnly = false, + creatableTopics, + Map.empty, + controllerMutationQuota, + topicErrors.set + ) + + val creatableTopicResponses = Option(topicErrors.get) match { + case Some(errors) => + errors.toSeq.map { case (topic, apiError) => + val error = apiError.error match { + case Errors.TOPIC_ALREADY_EXISTS | Errors.REQUEST_TIMED_OUT => + // The timeout error is expected because we set timeout=0. This + // nevertheless indicates that the topic metadata was created + // successfully, so we return LEADER_NOT_AVAILABLE. + Errors.LEADER_NOT_AVAILABLE + case error => error + } + + new MetadataResponseTopic() + .setErrorCode(error.code) + .setName(topic) + .setIsInternal(Topic.isInternal(topic)) + } + + case None => + creatableTopics.keySet.toSeq.map { topic => + new MetadataResponseTopic() + .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code) + .setName(topic) + .setIsInternal(Topic.isInternal(topic)) + } + } + + creatableTopicResponses + } finally { + clearInflightRequests(creatableTopics) + } + } + + private def sendCreateTopicRequest( + creatableTopics: Map[String, CreatableTopic] + ): Seq[MetadataResponseTopic] = { + val topicsToCreate = new CreateTopicsRequestData.CreatableTopicCollection(creatableTopics.size) + topicsToCreate.addAll(creatableTopics.values.asJavaCollection) + + val createTopicsRequest = new CreateTopicsRequest.Builder( + new CreateTopicsRequestData() + .setTimeoutMs(config.requestTimeoutMs) + .setTopics(topicsToCreate) + ) + + channelManager.get.sendRequest(createTopicsRequest, new ControllerRequestCompletionHandler { + override def onTimeout(): Unit = { + debug(s"Auto topic creation timed out for ${creatableTopics.keys}.") + clearInflightRequests(creatableTopics) + } + + override def onComplete(response: ClientResponse): Unit = { + debug(s"Auto topic creation completed for ${creatableTopics.keys}.") + clearInflightRequests(creatableTopics) + } + }) + + val creatableTopicResponses = creatableTopics.keySet.toSeq.map { topic => + new MetadataResponseTopic() + .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code) + .setName(topic) + .setIsInternal(Topic.isInternal(topic)) + } + + info(s"Sent auto-creation request for ${creatableTopics.keys} to the active controller.") + creatableTopicResponses + } + + private def clearInflightRequests(creatableTopics: Map[String, CreatableTopic]): Unit = { + creatableTopics.keySet.foreach(inflightTopics.remove) + debug(s"Cleared inflight topic creation state for $creatableTopics") + } + + private def creatableTopic(topic: String): CreatableTopic = { + topic match { + case GROUP_METADATA_TOPIC_NAME => + new CreatableTopic() + .setName(topic) + .setNumPartitions(config.offsetsTopicPartitions) + .setReplicationFactor(config.offsetsTopicReplicationFactor) + .setConfigs(convertToTopicConfigCollections(groupCoordinator.offsetsTopicConfigs)) + case TRANSACTION_STATE_TOPIC_NAME => + new CreatableTopic() + .setName(topic) + .setNumPartitions(config.transactionTopicPartitions) + .setReplicationFactor(config.transactionTopicReplicationFactor) + .setConfigs(convertToTopicConfigCollections( + txnCoordinator.transactionTopicConfigs)) + case topicName => + new CreatableTopic() + .setName(topicName) + .setNumPartitions(config.numPartitions) + .setReplicationFactor(config.defaultReplicationFactor.shortValue) + } + } + + private def convertToTopicConfigCollections(config: Properties): CreateableTopicConfigCollection = { + val topicConfigs = new CreateableTopicConfigCollection() + config.forEach { + case (name, value) => + topicConfigs.add(new CreateableTopicConfig() + .setName(name.toString) + .setValue(value.toString)) + } + topicConfigs + } + + private def isValidTopicName(topic: String): Boolean = { + try { + Topic.validate(topic) + true + } catch { + case _: InvalidTopicException => + false + } + } + + private def filterCreatableTopics( + topics: Set[String] + ): (Map[String, CreatableTopic], Seq[MetadataResponseTopic]) = { + + val aliveBrokers = metadataCache.getAliveBrokers + val creatableTopics = mutable.Map.empty[String, CreatableTopic] + val uncreatableTopics = mutable.Buffer.empty[MetadataResponseTopic] + + topics.foreach { topic => + // Attempt basic topic validation before sending any requests to the controller. + val validationError: Option[Errors] = if (!isValidTopicName(topic)) { + Some(Errors.INVALID_TOPIC_EXCEPTION) + } else if (!hasEnoughLiveBrokers(topic, aliveBrokers)) { + Some(Errors.INVALID_REPLICATION_FACTOR) + } else if (!inflightTopics.add(topic)) { + Some(Errors.UNKNOWN_TOPIC_OR_PARTITION) + } else { + None + } + + validationError match { + case Some(error) => + uncreatableTopics += new MetadataResponseTopic() + .setErrorCode(error.code) + .setName(topic) + .setIsInternal(Topic.isInternal(topic)) + case None => + creatableTopics.put(topic, creatableTopic(topic)) + } + } + + (creatableTopics, uncreatableTopics) + } + + private def hasEnoughLiveBrokers( + topicName: String, + aliveBrokers: Seq[MetadataBroker] + ): Boolean = { + val (replicationFactor, replicationFactorConfig) = topicName match { + case GROUP_METADATA_TOPIC_NAME => + (config.offsetsTopicReplicationFactor.intValue, KafkaConfig.OffsetsTopicReplicationFactorProp) + + case TRANSACTION_STATE_TOPIC_NAME => + (config.transactionTopicReplicationFactor.intValue, KafkaConfig.TransactionsTopicReplicationFactorProp) + + case _ => + (config.defaultReplicationFactor, KafkaConfig.DefaultReplicationFactorProp) + } + + if (aliveBrokers.size < replicationFactor) { + error(s"Number of alive brokers '${aliveBrokers.size}' does not meet the required replication factor " + + s"'$replicationFactor' for auto creation of topic '$topicName' which is configured by $replicationFactorConfig. " + + "This error can be ignored if the cluster is starting up and not all brokers are up yet.") + false + } else { + true + } + } + +} diff --git a/core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala b/core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala index 56ea6cbd8b43b..1e7af76aeb457 100644 --- a/core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala +++ b/core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala @@ -133,7 +133,6 @@ class BrokerToControllerChannelManagerImpl( def shutdown(): Unit = { requestThread.shutdown() - requestThread.awaitShutdown() info(s"Broker to controller channel manager for $channelName shutdown") } diff --git a/core/src/main/scala/kafka/server/ForwardingManager.scala b/core/src/main/scala/kafka/server/ForwardingManager.scala index c90387895cee3..b77bd3456359e 100644 --- a/core/src/main/scala/kafka/server/ForwardingManager.scala +++ b/core/src/main/scala/kafka/server/ForwardingManager.scala @@ -158,5 +158,4 @@ class ForwardingManagerImpl( request.getErrorResponse(Errors.UNKNOWN_SERVER_ERROR.exception) } } - } diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index 406749925857c..938c401f941df 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/core/src/main/scala/kafka/server/KafkaApis.scala @@ -21,11 +21,11 @@ import java.lang.{Long => JLong} import java.net.{InetAddress, UnknownHostException} import java.nio.ByteBuffer import java.util -import java.util.{Collections, Optional} import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.atomic.AtomicInteger +import java.util.{Collections, Optional} -import kafka.admin.{AdminUtils, RackAwareMode} +import kafka.admin.AdminUtils import kafka.api.{ApiVersion, ElectLeadersRequestOps, KAFKA_0_11_0_IV0, KAFKA_2_3_IV0} import kafka.common.OffsetAndMetadata import kafka.controller.ReplicaAssignment @@ -36,32 +36,33 @@ import kafka.message.ZStdCompressionCodec import kafka.network.RequestChannel import kafka.security.authorizer.AuthorizerUtils import kafka.server.QuotaFactory.{QuotaManagers, UnboundedQuota} -import kafka.utils.{CoreUtils, Logging} import kafka.utils.Implicits._ -import org.apache.kafka.clients.admin.{AlterConfigOp, ConfigEntry} +import kafka.utils.{CoreUtils, Logging} import org.apache.kafka.clients.admin.AlterConfigOp.OpType -import org.apache.kafka.common.acl.{AclBinding, AclOperation} +import org.apache.kafka.clients.admin.{AlterConfigOp, ConfigEntry} import org.apache.kafka.common.acl.AclOperation._ +import org.apache.kafka.common.acl.{AclBinding, AclOperation} import org.apache.kafka.common.config.ConfigResource import org.apache.kafka.common.errors._ -import org.apache.kafka.common.internals.{FatalExitError, Topic} import org.apache.kafka.common.internals.Topic.{GROUP_METADATA_TOPIC_NAME, TRANSACTION_STATE_TOPIC_NAME, isInternal} -import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic +import org.apache.kafka.common.internals.{FatalExitError, Topic} +import org.apache.kafka.common.message.AlterConfigsResponseData.AlterConfigsResourceResponse +import org.apache.kafka.common.message.AlterPartitionReassignmentsResponseData.{ReassignablePartitionResponse, ReassignableTopicResponse} +import org.apache.kafka.common.message.CreateAclsResponseData.AclCreationResult import org.apache.kafka.common.message.CreatePartitionsResponseData.CreatePartitionsTopicResult -import org.apache.kafka.common.message.{AddOffsetsToTxnResponseData, AlterClientQuotasResponseData, AlterConfigsResponseData, AlterPartitionReassignmentsResponseData, AlterReplicaLogDirsResponseData, ApiVersionsResponseData, CreateAclsResponseData, CreatePartitionsResponseData, CreateTopicsResponseData, DeleteAclsResponseData, DeleteGroupsResponseData, DeleteRecordsResponseData, DeleteTopicsResponseData, DescribeAclsResponseData, DescribeClientQuotasResponseData, DescribeClusterResponseData, DescribeConfigsResponseData, DescribeGroupsResponseData, DescribeLogDirsResponseData, DescribeProducersResponseData, EndTxnResponseData, ExpireDelegationTokenResponseData, FindCoordinatorResponseData, HeartbeatResponseData, InitProducerIdResponseData, JoinGroupResponseData, LeaveGroupResponseData, ListGroupsResponseData, ListOffsetsResponseData, ListPartitionReassignmentsResponseData, MetadataResponseData, OffsetCommitRequestData, OffsetCommitResponseData, OffsetDeleteResponseData, OffsetForLeaderEpochResponseData, RenewDelegationTokenResponseData, SaslAuthenticateResponseData, SaslHandshakeResponseData, StopReplicaResponseData, SyncGroupResponseData, UpdateMetadataResponseData} +import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic import org.apache.kafka.common.message.CreateTopicsResponseData.{CreatableTopicResult, CreatableTopicResultCollection} import org.apache.kafka.common.message.DeleteGroupsResponseData.{DeletableGroupResult, DeletableGroupResultCollection} -import org.apache.kafka.common.message.AlterPartitionReassignmentsResponseData.{ReassignablePartitionResponse, ReassignableTopicResponse} -import org.apache.kafka.common.message.CreateAclsResponseData.AclCreationResult -import org.apache.kafka.common.message.DeleteTopicsResponseData.{DeletableTopicResult, DeletableTopicResultCollection} import org.apache.kafka.common.message.DeleteRecordsResponseData.{DeleteRecordsPartitionResult, DeleteRecordsTopicResult} -import org.apache.kafka.common.message.ElectLeadersResponseData.PartitionResult -import org.apache.kafka.common.message.ElectLeadersResponseData.ReplicaElectionResult +import org.apache.kafka.common.message.DeleteTopicsResponseData.{DeletableTopicResult, DeletableTopicResultCollection} +import org.apache.kafka.common.message.ElectLeadersResponseData.{PartitionResult, ReplicaElectionResult} import org.apache.kafka.common.message.LeaveGroupResponseData.MemberResponse import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition import org.apache.kafka.common.message.ListOffsetsResponseData.{ListOffsetsPartitionResponse, ListOffsetsTopicResponse} +import org.apache.kafka.common.message.MetadataResponseData.{MetadataResponsePartition, MetadataResponseTopic} import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData.OffsetForLeaderTopic import org.apache.kafka.common.message.OffsetForLeaderEpochResponseData.{EpochEndOffset, OffsetForLeaderTopicResult, OffsetForLeaderTopicResultCollection} +import org.apache.kafka.common.message.{AddOffsetsToTxnResponseData, AlterClientQuotasResponseData, AlterConfigsResponseData, AlterPartitionReassignmentsResponseData, AlterReplicaLogDirsResponseData, ApiVersionsResponseData, CreateAclsResponseData, CreatePartitionsResponseData, CreateTopicsResponseData, DeleteAclsResponseData, DeleteGroupsResponseData, DeleteRecordsResponseData, DeleteTopicsResponseData, DescribeAclsResponseData, DescribeClientQuotasResponseData, DescribeClusterResponseData, DescribeConfigsResponseData, DescribeGroupsResponseData, DescribeLogDirsResponseData, DescribeProducersResponseData, EndTxnResponseData, ExpireDelegationTokenResponseData, FindCoordinatorResponseData, HeartbeatResponseData, InitProducerIdResponseData, JoinGroupResponseData, LeaveGroupResponseData, ListGroupsResponseData, ListOffsetsResponseData, ListPartitionReassignmentsResponseData, OffsetCommitRequestData, OffsetCommitResponseData, OffsetDeleteResponseData, OffsetForLeaderEpochResponseData, RenewDelegationTokenResponseData, SaslAuthenticateResponseData, SaslHandshakeResponseData, StopReplicaResponseData, SyncGroupResponseData, UpdateMetadataResponseData} import org.apache.kafka.common.metrics.Metrics import org.apache.kafka.common.network.{ClientInformation, ListenerName, Send} import org.apache.kafka.common.protocol.{ApiKeys, Errors} @@ -78,20 +79,17 @@ import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol} import org.apache.kafka.common.security.token.delegation.{DelegationToken, TokenInformation} import org.apache.kafka.common.utils.{ProducerIdAndEpoch, Time} import org.apache.kafka.common.{Node, TopicPartition, Uuid} -import org.apache.kafka.common.message.AlterConfigsResponseData.AlterConfigsResourceResponse -import org.apache.kafka.common.message.MetadataResponseData.{MetadataResponsePartition, MetadataResponseTopic} import org.apache.kafka.server.authorizer._ -import scala.compat.java8.OptionConverters._ -import scala.jdk.CollectionConverters._ +import scala.annotation.nowarn import scala.collection.mutable.ArrayBuffer import scala.collection.{Map, Seq, Set, immutable, mutable} +import scala.compat.java8.OptionConverters._ +import scala.jdk.CollectionConverters._ import scala.util.{Failure, Success, Try} import kafka.coordinator.group.GroupOverview import kafka.server.metadata.ConfigRepository -import scala.annotation.nowarn - /** * Logic to handle the various Kafka requests */ @@ -100,6 +98,7 @@ class KafkaApis(val requestChannel: RequestChannel, val replicaManager: ReplicaManager, val groupCoordinator: GroupCoordinator, val txnCoordinator: TransactionCoordinator, + val autoTopicCreationManager: AutoTopicCreationManager, val brokerId: Int, val config: KafkaConfig, val configRepository: ConfigRepository, @@ -1087,26 +1086,9 @@ class KafkaApis(val requestChannel: RequestChannel, (responseTopics ++ unauthorizedResponseStatus).toList } - private def createTopic(topic: String, - numPartitions: Int, - replicationFactor: Int, - properties: util.Properties = new util.Properties()): MetadataResponseTopic = { - // the below will be replaced once KAFKA-9751 is implemented - val zkSupport = metadataSupport.requireZkOrThrow(KafkaApis.notYetSupported("Auto topic creation")) - try { - zkSupport.adminZkClient.createTopic(topic, numPartitions, replicationFactor, properties, RackAwareMode.Safe, config.usesTopicId) - info("Auto creation of topic %s with %d partitions and replication factor %d is successful" - .format(topic, numPartitions, replicationFactor)) - metadataResponseTopic(Errors.LEADER_NOT_AVAILABLE, topic, isInternal(topic), util.Collections.emptyList()) - } catch { - case _: TopicExistsException => // let it go, possibly another broker created this topic - metadataResponseTopic(Errors.LEADER_NOT_AVAILABLE, topic, isInternal(topic), util.Collections.emptyList()) - case ex: Throwable => // Catch all to prevent unhandled errors - metadataResponseTopic(Errors.forException(ex), topic, isInternal(topic), util.Collections.emptyList()) - } - } - - private def metadataResponseTopic(error: Errors, topic: String, isInternal: Boolean, + private def metadataResponseTopic(error: Errors, + topic: String, + isInternal: Boolean, partitionData: util.List[MetadataResponsePartition]): MetadataResponseTopic = { new MetadataResponseTopic() .setErrorCode(error.code) @@ -1115,82 +1097,48 @@ class KafkaApis(val requestChannel: RequestChannel, .setPartitions(partitionData) } - private def createInternalTopic(topic: String): MetadataResponseTopic = { - if (topic == null) - throw new IllegalArgumentException("topic must not be null") - - val aliveBrokers = metadataCache.getAliveBrokers - - topic match { - case GROUP_METADATA_TOPIC_NAME => - if (aliveBrokers.size < config.offsetsTopicReplicationFactor) { - error(s"Number of alive brokers '${aliveBrokers.size}' does not meet the required replication factor " + - s"'${config.offsetsTopicReplicationFactor}' for the offsets topic (configured via " + - s"'${KafkaConfig.OffsetsTopicReplicationFactorProp}'). This error can be ignored if the cluster is starting up " + - s"and not all brokers are up yet.") - metadataResponseTopic(Errors.COORDINATOR_NOT_AVAILABLE, topic, true, util.Collections.emptyList()) - } else { - createTopic(topic, config.offsetsTopicPartitions, config.offsetsTopicReplicationFactor.toInt, - groupCoordinator.offsetsTopicConfigs) - } - case TRANSACTION_STATE_TOPIC_NAME => - if (aliveBrokers.size < config.transactionTopicReplicationFactor) { - error(s"Number of alive brokers '${aliveBrokers.size}' does not meet the required replication factor " + - s"'${config.transactionTopicReplicationFactor}' for the transactions state topic (configured via " + - s"'${KafkaConfig.TransactionsTopicReplicationFactorProp}'). This error can be ignored if the cluster is starting up " + - s"and not all brokers are up yet.") - metadataResponseTopic(Errors.COORDINATOR_NOT_AVAILABLE, topic, true, util.Collections.emptyList()) - } else { - createTopic(topic, config.transactionTopicPartitions, config.transactionTopicReplicationFactor.toInt, - txnCoordinator.transactionTopicConfigs) - } - case _ => throw new IllegalArgumentException(s"Unexpected internal topic name: $topic") - } - } - - private def getOrCreateInternalTopic(topic: String, listenerName: ListenerName): MetadataResponseData.MetadataResponseTopic = { - val topicMetadata = metadataCache.getTopicMetadata(Set(topic), listenerName) - topicMetadata.headOption.getOrElse(createInternalTopic(topic)) - } - - private def getTopicMetadata(allowAutoTopicCreation: Boolean, isFetchAllMetadata: Boolean, - topics: Set[String], listenerName: ListenerName, - errorUnavailableEndpoints: Boolean, - errorUnavailableListeners: Boolean): Seq[MetadataResponseTopic] = { + private def getTopicMetadata( + request: RequestChannel.Request, + fetchAllTopics: Boolean, + allowAutoTopicCreation: Boolean, + topics: Set[String], + listenerName: ListenerName, + errorUnavailableEndpoints: Boolean, + errorUnavailableListeners: Boolean + ): Seq[MetadataResponseTopic] = { val topicResponses = metadataCache.getTopicMetadata(topics, listenerName, - errorUnavailableEndpoints, errorUnavailableListeners) + errorUnavailableEndpoints, errorUnavailableListeners) - if (topics.isEmpty || topicResponses.size == topics.size) { + if (topics.isEmpty || topicResponses.size == topics.size || fetchAllTopics) { topicResponses } else { - val nonExistentTopics = topics.diff(topicResponses.map(_.name).toSet) - val responsesForNonExistentTopics = nonExistentTopics.flatMap { topic => - if (isInternal(topic)) { - val topicMetadata = createInternalTopic(topic) - Some( - if (topicMetadata.errorCode == Errors.COORDINATOR_NOT_AVAILABLE.code) - metadataResponseTopic(Errors.INVALID_REPLICATION_FACTOR, topic, true, util.Collections.emptyList()) - else - topicMetadata + val nonExistingTopics = topics.diff(topicResponses.map(_.name).toSet) + val nonExistingTopicResponses = if (allowAutoTopicCreation) { + val controllerMutationQuota = quotas.controllerMutation.newPermissiveQuotaFor(request) + autoTopicCreationManager.createTopics(nonExistingTopics, controllerMutationQuota) + } else { + nonExistingTopics.map { topic => + val error = try { + Topic.validate(topic) + Errors.UNKNOWN_TOPIC_OR_PARTITION + } catch { + case _: InvalidTopicException => + Errors.INVALID_TOPIC_EXCEPTION + } + + metadataResponseTopic( + error, + topic, + Topic.isInternal(topic), + util.Collections.emptyList() ) - } else if (isFetchAllMetadata) { - // A metadata request for all topics should never result in topic auto creation, but a topic may be deleted - // in between the creation of the topics parameter and topicResponses, so make sure to return None for this case. - None - } else if (allowAutoTopicCreation && config.autoCreateTopicsEnable) { - Some(createTopic(topic, config.numPartitions, config.defaultReplicationFactor)) - } else { - Some(metadataResponseTopic(Errors.UNKNOWN_TOPIC_OR_PARTITION, topic, false, util.Collections.emptyList())) } } - topicResponses ++ responsesForNonExistentTopics + topicResponses ++ nonExistingTopicResponses } } - /** - * Handle a topic metadata request - */ def handleTopicMetadataRequest(request: RequestChannel.Request): Unit = { val metadataRequest = request.body[MetadataRequest] val requestVersion = request.header.apiVersion @@ -1220,7 +1168,6 @@ class KafkaApis(val requestChannel: RequestChannel, val unauthorizedForCreateTopicMetadata = unauthorizedForCreateTopics.map(topic => metadataResponseTopic(Errors.TOPIC_AUTHORIZATION_FAILED, topic, isInternal(topic), util.Collections.emptyList())) - // do not disclose the existence of topics unauthorized for Describe, so we've not even checked if they exist or not val unauthorizedForDescribeTopicMetadata = // In case of all topics, don't include topics unauthorized for Describe @@ -1236,19 +1183,10 @@ class KafkaApis(val requestChannel: RequestChannel, // In versions 5 and below, we returned LEADER_NOT_AVAILABLE if a matching listener was not found on the leader. // From version 6 onwards, we return LISTENER_NOT_FOUND to enable diagnosis of configuration errors. val errorUnavailableListeners = requestVersion >= 6 - val topicMetadata = - if (authorizedTopics.isEmpty) - Seq.empty[MetadataResponseTopic] - else { - getTopicMetadata( - metadataRequest.allowAutoTopicCreation, - metadataRequest.isAllTopics, - authorizedTopics, - request.context.listenerName, - errorUnavailableEndpoints, - errorUnavailableListeners - ) - } + + val allowAutoCreation = config.autoCreateTopicsEnable && metadataRequest.allowAutoTopicCreation && !metadataRequest.isAllTopics + val topicMetadata = getTopicMetadata(request, metadataRequest.isAllTopics, allowAutoCreation, authorizedTopics, + request.context.listenerName, errorUnavailableEndpoints, errorUnavailableListeners) var clusterAuthorizedOperations = Int.MinValue // Default value in the schema if (requestVersion >= 8) { @@ -1264,9 +1202,12 @@ class KafkaApis(val requestChannel: RequestChannel, // get topic authorized operations if (metadataRequest.data.includeTopicAuthorizedOperations) { - topicMetadata.foreach { topicData => - topicData.setTopicAuthorizedOperations(authHelper.authorizedOperations(request, new Resource(ResourceType.TOPIC, topicData.name))) + def setTopicAuthorizedOperations(topicMetadata: Seq[MetadataResponseTopic]): Unit = { + topicMetadata.foreach { topicData => + topicData.setTopicAuthorizedOperations(authHelper.authorizedOperations(request, new Resource(ResourceType.TOPIC, topicData.name))) + } } + setTopicAuthorizedOperations(topicMetadata) } } @@ -1378,55 +1319,58 @@ class KafkaApis(val requestChannel: RequestChannel, !authHelper.authorize(request.context, DESCRIBE, TRANSACTIONAL_ID, findCoordinatorRequest.data.key)) requestHelper.sendErrorResponseMaybeThrottle(request, Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED.exception) else { - // get metadata (and create the topic if necessary) - val (partition, topicMetadata) = CoordinatorType.forId(findCoordinatorRequest.data.keyType) match { + val (partition, internalTopicName) = CoordinatorType.forId(findCoordinatorRequest.data.keyType) match { case CoordinatorType.GROUP => - val partition = groupCoordinator.partitionFor(findCoordinatorRequest.data.key) - val metadata = getOrCreateInternalTopic(GROUP_METADATA_TOPIC_NAME, request.context.listenerName) - (partition, metadata) + (groupCoordinator.partitionFor(findCoordinatorRequest.data.key), GROUP_METADATA_TOPIC_NAME) case CoordinatorType.TRANSACTION => - val partition = txnCoordinator.partitionFor(findCoordinatorRequest.data.key) - val metadata = getOrCreateInternalTopic(TRANSACTION_STATE_TOPIC_NAME, request.context.listenerName) - (partition, metadata) + (txnCoordinator.partitionFor(findCoordinatorRequest.data.key), TRANSACTION_STATE_TOPIC_NAME) + } - case _ => - throw new InvalidRequestException("Unknown coordinator type in FindCoordinator request") + val topicMetadata = metadataCache.getTopicMetadata(Set(internalTopicName), request.context.listenerName) + def createFindCoordinatorResponse(error: Errors, + node: Node, + requestThrottleMs: Int): FindCoordinatorResponse = { + new FindCoordinatorResponse( + new FindCoordinatorResponseData() + .setErrorCode(error.code) + .setErrorMessage(error.message()) + .setNodeId(node.id) + .setHost(node.host) + .setPort(node.port) + .setThrottleTimeMs(requestThrottleMs)) } - def createResponse(requestThrottleMs: Int): AbstractResponse = { - def createFindCoordinatorResponse(error: Errors, node: Node): FindCoordinatorResponse = { - new FindCoordinatorResponse( - new FindCoordinatorResponseData() - .setErrorCode(error.code) - .setErrorMessage(error.message) - .setNodeId(node.id) - .setHost(node.host) - .setPort(node.port) - .setThrottleTimeMs(requestThrottleMs)) - } - val responseBody = if (topicMetadata.errorCode != Errors.NONE.code) { - createFindCoordinatorResponse(Errors.COORDINATOR_NOT_AVAILABLE, Node.noNode) - } else { - val coordinatorEndpoint = topicMetadata.partitions.asScala - .find(_.partitionIndex == partition) - .filter(_.leaderId != MetadataResponse.NO_LEADER_ID) - .flatMap(metadata => metadataCache.getAliveBroker(metadata.leaderId)) - .flatMap(_.endpoints.get(request.context.listenerName.value())) - .filterNot(_.isEmpty) - - coordinatorEndpoint match { - case Some(endpoint) => - createFindCoordinatorResponse(Errors.NONE, endpoint) - case _ => - createFindCoordinatorResponse(Errors.COORDINATOR_NOT_AVAILABLE, Node.noNode) + if (topicMetadata.headOption.isEmpty) { + val controllerMutationQuota = quotas.controllerMutation.newPermissiveQuotaFor(request) + autoTopicCreationManager.createTopics(Seq(internalTopicName).toSet, controllerMutationQuota) + requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs => createFindCoordinatorResponse( + Errors.COORDINATOR_NOT_AVAILABLE, Node.noNode, requestThrottleMs)) + } else { + def createResponse(requestThrottleMs: Int): AbstractResponse = { + val responseBody = if (topicMetadata.head.errorCode != Errors.NONE.code) { + createFindCoordinatorResponse(Errors.COORDINATOR_NOT_AVAILABLE, Node.noNode, requestThrottleMs) + } else { + val coordinatorEndpoint = topicMetadata.head.partitions.asScala + .find(_.partitionIndex == partition) + .filter(_.leaderId != MetadataResponse.NO_LEADER_ID) + .flatMap(metadata => metadataCache.getAliveBroker(metadata.leaderId)) + .flatMap(_.endpoints.get(request.context.listenerName.value())) + .filterNot(_.isEmpty) + + coordinatorEndpoint match { + case Some(endpoint) => + createFindCoordinatorResponse(Errors.NONE, endpoint, requestThrottleMs) + case _ => + createFindCoordinatorResponse(Errors.COORDINATOR_NOT_AVAILABLE, Node.noNode, requestThrottleMs) + } } + trace("Sending FindCoordinator response %s for correlation id %d to client %s." + .format(responseBody, request.header.correlationId, request.header.clientId)) + responseBody } - trace("Sending FindCoordinator response %s for correlation id %d to client %s." - .format(responseBody, request.header.correlationId, request.header.clientId)) - responseBody + requestHelper.sendResponseMaybeThrottle(request, createResponse) } - requestHelper.sendResponseMaybeThrottle(request, createResponse) } } diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala index f22ef54f868f7..dcd0ff0cf64e2 100755 --- a/core/src/main/scala/kafka/server/KafkaServer.scala +++ b/core/src/main/scala/kafka/server/KafkaServer.scala @@ -132,6 +132,8 @@ class KafkaServer( var forwardingManager: Option[ForwardingManager] = None + var autoTopicCreationManager: AutoTopicCreationManager = null + var alterIsrManager: AlterIsrManager = null var kafkaScheduler: KafkaScheduler = null @@ -294,6 +296,7 @@ class KafkaServer( kafkaController = new KafkaController(config, zkClient, time, metrics, brokerInfo, brokerEpoch, tokenManager, brokerFeatures, featureCache, threadNamePrefix) kafkaController.startup() + /* start forwarding manager */ if (enableForwarding) { this.forwardingManager = Some(ForwardingManager( config, @@ -319,6 +322,21 @@ class KafkaServer( transactionCoordinator.startup( () => zkClient.getTopicPartitionCount(Topic.TRANSACTION_STATE_TOPIC_NAME).getOrElse(config.transactionTopicPartitions)) + /* start auto topic creation manager */ + this.autoTopicCreationManager = AutoTopicCreationManager( + config, + metadataCache, + time, + metrics, + threadNamePrefix, + adminManager, + kafkaController, + groupCoordinator, + transactionCoordinator, + enableForwarding + ) + autoTopicCreationManager.start() + /* Get the authorizer and initialize it if one is specified.*/ authorizer = config.authorizer authorizer.foreach(_.configure(config.originals)) @@ -340,7 +358,7 @@ class KafkaServer( /* start processing requests */ val zkSupport = ZkSupport(adminManager, kafkaController, zkClient, forwardingManager) dataPlaneRequestProcessor = new KafkaApis(socketServer.dataPlaneRequestChannel, zkSupport, replicaManager, groupCoordinator, transactionCoordinator, - config.brokerId, config, configRepository, metadataCache, metrics, authorizer, quotaManagers, + autoTopicCreationManager, config.brokerId, config, configRepository, metadataCache, metrics, authorizer, quotaManagers, fetchManager, brokerTopicStats, clusterId, time, tokenManager, brokerFeatures, featureCache) dataPlaneRequestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.dataPlaneRequestChannel, dataPlaneRequestProcessor, time, @@ -348,7 +366,7 @@ class KafkaServer( socketServer.controlPlaneRequestChannelOpt.foreach { controlPlaneRequestChannel => controlPlaneRequestProcessor = new KafkaApis(controlPlaneRequestChannel, zkSupport, replicaManager, groupCoordinator, transactionCoordinator, - config.brokerId, config, configRepository, metadataCache, metrics, authorizer, quotaManagers, + autoTopicCreationManager, config.brokerId, config, configRepository, metadataCache, metrics, authorizer, quotaManagers, fetchManager, brokerTopicStats, clusterId, time, tokenManager, brokerFeatures, featureCache) controlPlaneRequestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.controlPlaneRequestChannelOpt.get, controlPlaneRequestProcessor, time, @@ -674,8 +692,10 @@ class KafkaServer( if (alterIsrManager != null) CoreUtils.swallow(alterIsrManager.shutdown(), this) - if (forwardingManager != null) - CoreUtils.swallow(forwardingManager.foreach(_.shutdown()), this) + CoreUtils.swallow(forwardingManager.foreach(_.shutdown()), this) + + if (autoTopicCreationManager != null) + CoreUtils.swallow(autoTopicCreationManager.shutdown(), this) if (logManager != null) CoreUtils.swallow(logManager.shutdown(), this) diff --git a/core/src/test/scala/integration/kafka/api/ConsumerTopicCreationTest.scala b/core/src/test/scala/integration/kafka/api/ConsumerTopicCreationTest.scala index e68ce2ff60ca1..9335eac42be17 100644 --- a/core/src/test/scala/integration/kafka/api/ConsumerTopicCreationTest.scala +++ b/core/src/test/scala/integration/kafka/api/ConsumerTopicCreationTest.scala @@ -17,6 +17,12 @@ package kafka.api +import java.lang.{Boolean => JBoolean} +import java.time.Duration +import java.util +import java.util.Collections + +import kafka.api import kafka.server.KafkaConfig import kafka.utils.TestUtils import org.apache.kafka.clients.admin.NewTopic @@ -26,15 +32,11 @@ import org.junit.jupiter.api.Assertions._ import org.junit.jupiter.params.ParameterizedTest import org.junit.jupiter.params.provider.{Arguments, MethodSource} -import java.lang.{Boolean => JBoolean} -import java.time.Duration -import java.util -import java.util.Collections - /** * Tests behavior of specifying auto topic creation configuration for the consumer and broker */ class ConsumerTopicCreationTest { + @ParameterizedTest @MethodSource(Array("parameters")) def testAutoTopicCreation(brokerAutoTopicCreationEnable: JBoolean, consumerAllowAutoCreateTopics: JBoolean): Unit = { @@ -42,10 +44,26 @@ class ConsumerTopicCreationTest { testCase.setUp() try testCase.test() finally testCase.tearDown() } + + @ParameterizedTest + @MethodSource(Array("parameters")) + def testAutoTopicCreationWithForwarding(brokerAutoTopicCreationEnable: JBoolean, consumerAllowAutoCreateTopics: JBoolean): Unit = { + val testCase = new api.ConsumerTopicCreationTest.TestCaseWithForwarding(brokerAutoTopicCreationEnable, consumerAllowAutoCreateTopics) + testCase.setUp() + try testCase.test() finally testCase.tearDown() + } } object ConsumerTopicCreationTest { + private class TestCaseWithForwarding(brokerAutoTopicCreationEnable: JBoolean, consumerAllowAutoCreateTopics: JBoolean) + extends TestCase(brokerAutoTopicCreationEnable, consumerAllowAutoCreateTopics) { + + override protected def brokerCount: Int = 3 + + override def enableForwarding: Boolean = true + } + private class TestCase(brokerAutoTopicCreationEnable: JBoolean, consumerAllowAutoCreateTopics: JBoolean) extends IntegrationTestHarness { private val topic_1 = "topic-1" private val topic_2 = "topic-2" diff --git a/core/src/test/scala/integration/kafka/api/MetricsTest.scala b/core/src/test/scala/integration/kafka/api/MetricsTest.scala index 64c40993f8165..068ff103dbb4f 100644 --- a/core/src/test/scala/integration/kafka/api/MetricsTest.scala +++ b/core/src/test/scala/integration/kafka/api/MetricsTest.scala @@ -23,7 +23,7 @@ import org.apache.kafka.clients.consumer.KafkaConsumer import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord} import org.apache.kafka.common.{Metric, MetricName, TopicPartition} import org.apache.kafka.common.config.SaslConfigs -import org.apache.kafka.common.errors.InvalidTopicException +import org.apache.kafka.common.errors.{InvalidTopicException, UnknownTopicOrPartitionException} import org.apache.kafka.common.network.ListenerName import org.apache.kafka.common.security.auth.SecurityProtocol import org.apache.kafka.common.security.authenticator.TestJaasConfig @@ -42,7 +42,7 @@ class MetricsTest extends IntegrationTestHarness with SaslSetup { private val kafkaServerJaasEntryName = s"${listenerName.value.toLowerCase(Locale.ROOT)}.${JaasTestUtils.KafkaServerContextName}" this.serverConfig.setProperty(KafkaConfig.ZkEnableSecureAclsProp, "false") - this.serverConfig.setProperty(KafkaConfig.AutoCreateTopicsEnableDoc, "false") + this.serverConfig.setProperty(KafkaConfig.AutoCreateTopicsEnableProp, "false") this.producerConfig.setProperty(ProducerConfig.LINGER_MS_CONFIG, "10") // intentionally slow message down conversion via gzip compression to ensure we can measure the time it takes this.producerConfig.setProperty(ProducerConfig.COMPRESSION_TYPE_CONFIG, "gzip") @@ -226,8 +226,8 @@ class MetricsTest extends IntegrationTestHarness with SaslSetup { val errorMetricPrefix = "kafka.network:type=RequestMetrics,name=ErrorsPerSec" verifyYammerMetricRecorded(s"$errorMetricPrefix,request=Metadata,error=NONE") + val consumer = createConsumer() try { - val consumer = createConsumer() consumer.partitionsFor("12{}!") } catch { case _: InvalidTopicException => // expected @@ -239,10 +239,12 @@ class MetricsTest extends IntegrationTestHarness with SaslSetup { assertEquals(startErrorMetricCount + 1, currentErrorMetricCount) assertTrue(currentErrorMetricCount < 10, s"Too many error metrics $currentErrorMetricCount") - // Verify that error metric is updated with producer acks=0 when no response is sent - val producer = createProducer() - sendRecords(producer, numRecords = 1, recordSize = 100, new TopicPartition("non-existent", 0)) - verifyYammerMetricRecorded(s"$errorMetricPrefix,request=Metadata,error=LEADER_NOT_AVAILABLE") + try { + consumer.partitionsFor("non-existing-topic") + } catch { + case _: UnknownTopicOrPartitionException => // expected + } + verifyYammerMetricRecorded(s"$errorMetricPrefix,request=Metadata,error=UNKNOWN_TOPIC_OR_PARTITION") } private def verifyKafkaMetric[T](name: String, metrics: java.util.Map[MetricName, _ <: Metric], entity: String, diff --git a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala index 5b09b288d7a06..d0b9084227c49 100644 --- a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala +++ b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala @@ -557,6 +557,8 @@ class PlaintextConsumerTest extends BaseConsumerTest { @Test def testPartitionsForAutoCreate(): Unit = { val consumer = createConsumer() + // First call would create the topic + consumer.partitionsFor("non-exist-topic") val partitions = consumer.partitionsFor("non-exist-topic") assertFalse(partitions.isEmpty) } diff --git a/core/src/test/scala/unit/kafka/server/AbstractCreateTopicsRequestTest.scala b/core/src/test/scala/unit/kafka/server/AbstractCreateTopicsRequestTest.scala index 9d21fc2b41647..b6036b7c5e830 100644 --- a/core/src/test/scala/unit/kafka/server/AbstractCreateTopicsRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/AbstractCreateTopicsRequestTest.scala @@ -100,7 +100,7 @@ abstract class AbstractCreateTopicsRequestTest extends BaseRequestTest { request.data.topics.forEach { topic => def verifyMetadata(socketServer: SocketServer) = { val metadata = sendMetadataRequest( - new MetadataRequest.Builder(List(topic.name()).asJava, true).build()).topicMetadata.asScala + new MetadataRequest.Builder(List(topic.name()).asJava, false).build()).topicMetadata.asScala val metadataForTopic = metadata.filter(_.topic == topic.name()).head val partitions = if (!topic.assignments().isEmpty) diff --git a/core/src/test/scala/unit/kafka/server/AbstractMetadataRequestTest.scala b/core/src/test/scala/unit/kafka/server/AbstractMetadataRequestTest.scala new file mode 100644 index 0000000000000..b14085137c31c --- /dev/null +++ b/core/src/test/scala/unit/kafka/server/AbstractMetadataRequestTest.scala @@ -0,0 +1,61 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package kafka.server + +import java.util.Properties + +import kafka.network.SocketServer +import kafka.utils.TestUtils +import org.apache.kafka.common.message.MetadataRequestData +import org.apache.kafka.common.protocol.Errors +import org.apache.kafka.common.requests.{MetadataRequest, MetadataResponse} +import org.junit.jupiter.api.Assertions.assertEquals + +abstract class AbstractMetadataRequestTest extends BaseRequestTest { + + override def brokerPropertyOverrides(properties: Properties): Unit = { + properties.setProperty(KafkaConfig.OffsetsTopicPartitionsProp, "1") + properties.setProperty(KafkaConfig.DefaultReplicationFactorProp, "2") + properties.setProperty(KafkaConfig.RackProp, s"rack/${properties.getProperty(KafkaConfig.BrokerIdProp)}") + } + + protected def requestData(topics: List[String], allowAutoTopicCreation: Boolean): MetadataRequestData = { + val data = new MetadataRequestData + if (topics == null) + data.setTopics(null) + else + topics.foreach(topic => + data.topics.add( + new MetadataRequestData.MetadataRequestTopic() + .setName(topic))) + + data.setAllowAutoTopicCreation(allowAutoTopicCreation) + data + } + + protected def sendMetadataRequest(request: MetadataRequest, destination: Option[SocketServer] = None): MetadataResponse = { + connectAndReceive[MetadataResponse](request, destination = destination.getOrElse(anySocketServer)) + } + + protected def checkAutoCreatedTopic(autoCreatedTopic: String, response: MetadataResponse): Unit = { + assertEquals(Errors.LEADER_NOT_AVAILABLE, response.errors.get(autoCreatedTopic)) + assertEquals(Some(servers.head.config.numPartitions), zkClient.getTopicPartitionCount(autoCreatedTopic)) + for (i <- 0 until servers.head.config.numPartitions) + TestUtils.waitForPartitionMetadata(servers, autoCreatedTopic, i) + } +} diff --git a/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala b/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala new file mode 100644 index 0000000000000..dc4dd06ef6115 --- /dev/null +++ b/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala @@ -0,0 +1,194 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package kafka.server + +import java.util.Properties + +import kafka.controller.KafkaController +import kafka.coordinator.group.GroupCoordinator +import kafka.coordinator.transaction.TransactionCoordinator +import kafka.utils.TestUtils +import kafka.utils.TestUtils.createBroker +import org.apache.kafka.common.internals.Topic.{GROUP_METADATA_TOPIC_NAME, TRANSACTION_STATE_TOPIC_NAME} +import org.apache.kafka.common.message.CreateTopicsRequestData +import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic +import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseTopic +import org.apache.kafka.common.protocol.Errors +import org.apache.kafka.common.requests._ +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.{BeforeEach, Test} +import org.mockito.ArgumentMatchers.any +import org.mockito.{ArgumentMatchers, Mockito} + +import scala.collection.{Map, Seq} + +class AutoTopicCreationManagerTest { + + private val requestTimeout = 100 + private var config: KafkaConfig = _ + private val metadataCache = Mockito.mock(classOf[MetadataCache]) + private val brokerToController = Mockito.mock(classOf[BrokerToControllerChannelManager]) + private val adminManager = Mockito.mock(classOf[ZkAdminManager]) + private val controller = Mockito.mock(classOf[KafkaController]) + private val groupCoordinator = Mockito.mock(classOf[GroupCoordinator]) + private val transactionCoordinator = Mockito.mock(classOf[TransactionCoordinator]) + private var autoTopicCreationManager: AutoTopicCreationManager = _ + + private val internalTopicPartitions = 2 + private val internalTopicReplicationFactor: Short = 2 + + @BeforeEach + def setup(): Unit = { + val props = TestUtils.createBrokerConfig(1, "localhost") + props.setProperty(KafkaConfig.RequestTimeoutMsProp, requestTimeout.toString) + + props.setProperty(KafkaConfig.OffsetsTopicReplicationFactorProp, internalTopicPartitions.toString) + props.setProperty(KafkaConfig.TransactionsTopicReplicationFactorProp, internalTopicPartitions.toString) + + props.setProperty(KafkaConfig.OffsetsTopicPartitionsProp, internalTopicReplicationFactor.toString) + props.setProperty(KafkaConfig.TransactionsTopicPartitionsProp, internalTopicReplicationFactor.toString) + + config = KafkaConfig.fromProps(props) + val aliveBrokers = Seq(createBroker(0, "host0", 0), createBroker(1, "host1", 1)) + + Mockito.reset(metadataCache, controller, brokerToController, groupCoordinator, transactionCoordinator) + + Mockito.when(metadataCache.getAliveBrokers).thenReturn(aliveBrokers) + } + + @Test + def testCreateOffsetTopic(): Unit = { + Mockito.when(groupCoordinator.offsetsTopicConfigs).thenReturn(new Properties) + testCreateTopic(GROUP_METADATA_TOPIC_NAME, true, internalTopicPartitions, internalTopicReplicationFactor) + } + + @Test + def testCreateTxnTopic(): Unit = { + Mockito.when(transactionCoordinator.transactionTopicConfigs).thenReturn(new Properties) + testCreateTopic(TRANSACTION_STATE_TOPIC_NAME, true, internalTopicPartitions, internalTopicReplicationFactor) + } + + @Test + def testCreateNonInternalTopic(): Unit = { + testCreateTopic("topic", false) + } + + private def testCreateTopic(topicName: String, + isInternal: Boolean, + numPartitions: Int = 1, + replicationFactor: Short = 1): Unit = { + autoTopicCreationManager = new DefaultAutoTopicCreationManager( + config, + metadataCache, + Some(brokerToController), + adminManager, + controller, + groupCoordinator, + transactionCoordinator) + + val topicsCollection = new CreateTopicsRequestData.CreatableTopicCollection + topicsCollection.add(getNewTopic(topicName, numPartitions, replicationFactor)) + val requestBody = new CreateTopicsRequest.Builder( + new CreateTopicsRequestData() + .setTopics(topicsCollection) + .setTimeoutMs(requestTimeout)) + + Mockito.when(controller.isActive).thenReturn(false) + + // Calling twice with the same topic will only trigger one forwarding. + createTopicAndVerifyResult(Errors.UNKNOWN_TOPIC_OR_PARTITION, topicName, isInternal) + createTopicAndVerifyResult(Errors.UNKNOWN_TOPIC_OR_PARTITION, topicName, isInternal) + + Mockito.verify(brokerToController).sendRequest( + ArgumentMatchers.eq(requestBody), + any(classOf[ControllerRequestCompletionHandler])) + } + + @Test + def testCreateTopicsWithForwardingDisabled(): Unit = { + autoTopicCreationManager = new DefaultAutoTopicCreationManager( + config, + metadataCache, + None, + adminManager, + controller, + groupCoordinator, + transactionCoordinator) + + val topicName = "topic" + + Mockito.when(controller.isActive).thenReturn(false) + + createTopicAndVerifyResult(Errors.UNKNOWN_TOPIC_OR_PARTITION, topicName, false) + + Mockito.verify(adminManager).createTopics( + ArgumentMatchers.eq(0), + ArgumentMatchers.eq(false), + ArgumentMatchers.eq(Map(topicName -> getNewTopic(topicName))), + ArgumentMatchers.eq(Map.empty), + any(classOf[ControllerMutationQuota]), + any(classOf[Map[String, ApiError] => Unit])) + } + + @Test + def testNotEnoughLiveBrokers(): Unit = { + val props = TestUtils.createBrokerConfig(1, "localhost") + props.setProperty(KafkaConfig.DefaultReplicationFactorProp, 3.toString) + config = KafkaConfig.fromProps(props) + + autoTopicCreationManager = new DefaultAutoTopicCreationManager( + config, + metadataCache, + Some(brokerToController), + adminManager, + controller, + groupCoordinator, + transactionCoordinator) + + val topicName = "topic" + + Mockito.when(controller.isActive).thenReturn(false) + + createTopicAndVerifyResult(Errors.INVALID_REPLICATION_FACTOR, topicName, false) + + Mockito.verify(brokerToController, Mockito.never()).sendRequest( + any(), + any(classOf[ControllerRequestCompletionHandler])) + } + + private def createTopicAndVerifyResult(error: Errors, + topicName: String, + isInternal: Boolean): Unit = { + val topicResponses = autoTopicCreationManager.createTopics( + Set(topicName), UnboundedControllerMutationQuota) + + val expectedResponses = Seq(new MetadataResponseTopic() + .setErrorCode(error.code()) + .setIsInternal(isInternal) + .setName(topicName)) + + assertEquals(expectedResponses, topicResponses) + } + + private def getNewTopic(topicName: String, numPartitions: Int = 1, replicationFactor: Short = 1): CreatableTopic = { + new CreatableTopic() + .setName(topicName) + .setNumPartitions(numPartitions) + .setReplicationFactor(replicationFactor) + } +} diff --git a/core/src/test/scala/unit/kafka/server/CreateTopicsRequestWithForwardingTest.scala b/core/src/test/scala/unit/kafka/server/CreateTopicsRequestWithForwardingTest.scala index a538cebcc7c11..f55aa34d0c4f9 100644 --- a/core/src/test/scala/unit/kafka/server/CreateTopicsRequestWithForwardingTest.scala +++ b/core/src/test/scala/unit/kafka/server/CreateTopicsRequestWithForwardingTest.scala @@ -34,5 +34,4 @@ class CreateTopicsRequestWithForwardingTest extends AbstractCreateTopicsRequestT // With forwarding enabled, request could be forwarded to the active controller. assertEquals(Map(Errors.NONE -> 1), response.errorCounts().asScala) } - } diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala index 68c26ce420a0d..88bf8ebc3affe 100644 --- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala +++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala @@ -62,6 +62,8 @@ import org.apache.kafka.common.quota.{ClientQuotaAlteration, ClientQuotaEntity} import org.apache.kafka.common.record.FileRecords.TimestampAndOffset import org.apache.kafka.common.record._ import org.apache.kafka.common.replica.ClientMetadata +import org.apache.kafka.common.requests.FindCoordinatorRequest.CoordinatorType +import org.apache.kafka.common.requests.MetadataResponse.TopicMetadata import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse import org.apache.kafka.common.requests.WriteTxnMarkersRequest.TxnMarkerEntry import org.apache.kafka.common.requests.{FetchMetadata => JFetchMetadata, _} @@ -90,6 +92,8 @@ class KafkaApisTest { private val txnCoordinator: TransactionCoordinator = EasyMock.createNiceMock(classOf[TransactionCoordinator]) private val controller: KafkaController = EasyMock.createNiceMock(classOf[KafkaController]) private val forwardingManager: ForwardingManager = EasyMock.createNiceMock(classOf[ForwardingManager]) + private val autoTopicCreationManager: AutoTopicCreationManager = EasyMock.createNiceMock(classOf[AutoTopicCreationManager]) + private val hostAddress: Array[Byte] = InetAddress.getByName("192.168.1.1").getAddress private val kafkaPrincipalSerde = new KafkaPrincipalSerde { override def serialize(principal: KafkaPrincipal): Array[Byte] = Utils.utf8(principal.toString) @@ -122,7 +126,8 @@ class KafkaApisTest { authorizer: Option[Authorizer] = None, enableForwarding: Boolean = false, configRepository: ConfigRepository = new CachedConfigRepository(), - raftSupport: Boolean = false): KafkaApis = { + raftSupport: Boolean = false, + overrideProperties: Map[String, String] = Map.empty): KafkaApis = { val brokerFeatures = BrokerFeatures.createDefault() val cache = new FinalizedFeatureCache(brokerFeatures) @@ -134,6 +139,7 @@ class KafkaApisTest { } else { TestUtils.createBrokerConfig(brokerId, "zk") } + overrideProperties.foreach( p => properties.put(p._1, p._2)) properties.put(KafkaConfig.InterBrokerProtocolVersionProp, interBrokerProtocolVersion.toString) properties.put(KafkaConfig.LogMessageFormatVersionProp, interBrokerProtocolVersion.toString) @@ -147,6 +153,7 @@ class KafkaApisTest { replicaManager, groupCoordinator, txnCoordinator, + autoTopicCreationManager, brokerId, new KafkaConfig(properties), configRepository, @@ -887,6 +894,223 @@ class KafkaApisTest { testForwardableAPI(ApiKeys.ALTER_USER_SCRAM_CREDENTIALS, requestBuilder) } + @Test + def testFindCoordinatorAutoTopicCreationForOffsetTopic(): Unit = { + testFindCoordinatorWithTopicCreation(CoordinatorType.GROUP) + } + + @Test + def testFindCoordinatorAutoTopicCreationForTxnTopic(): Unit = { + testFindCoordinatorWithTopicCreation(CoordinatorType.TRANSACTION) + } + + @Test + def testFindCoordinatorNotEnoughBrokersForOffsetTopic(): Unit = { + testFindCoordinatorWithTopicCreation(CoordinatorType.GROUP, hasEnoughLiveBrokers = false) + } + + @Test + def testFindCoordinatorNotEnoughBrokersForTxnTopic(): Unit = { + testFindCoordinatorWithTopicCreation(CoordinatorType.TRANSACTION, hasEnoughLiveBrokers = false) + } + + private def testFindCoordinatorWithTopicCreation(coordinatorType: CoordinatorType, + hasEnoughLiveBrokers: Boolean = true): Unit = { + val authorizer: Authorizer = EasyMock.niceMock(classOf[Authorizer]) + + val requestHeader = new RequestHeader(ApiKeys.FIND_COORDINATOR, ApiKeys.FIND_COORDINATOR.latestVersion, + clientId, 0) + + val numBrokersNeeded = 3 + + setupBrokerMetadata(hasEnoughLiveBrokers, numBrokersNeeded) + + val requestTimeout = 10 + val topicConfigOverride = mutable.Map.empty[String, String] + topicConfigOverride.put(KafkaConfig.RequestTimeoutMsProp, requestTimeout.toString) + + val groupId = "group" + val topicName = + coordinatorType match { + case CoordinatorType.GROUP => + topicConfigOverride.put(KafkaConfig.OffsetsTopicPartitionsProp, numBrokersNeeded.toString) + topicConfigOverride.put(KafkaConfig.OffsetsTopicReplicationFactorProp, numBrokersNeeded.toString) + EasyMock.expect(groupCoordinator.offsetsTopicConfigs).andReturn(new Properties) + authorizeResource(authorizer, AclOperation.DESCRIBE, ResourceType.GROUP, + groupId, AuthorizationResult.ALLOWED) + Topic.GROUP_METADATA_TOPIC_NAME + case CoordinatorType.TRANSACTION => + topicConfigOverride.put(KafkaConfig.TransactionsTopicPartitionsProp, numBrokersNeeded.toString) + topicConfigOverride.put(KafkaConfig.TransactionsTopicReplicationFactorProp, numBrokersNeeded.toString) + EasyMock.expect(txnCoordinator.transactionTopicConfigs).andReturn(new Properties) + authorizeResource(authorizer, AclOperation.DESCRIBE, ResourceType.TRANSACTIONAL_ID, + groupId, AuthorizationResult.ALLOWED) + Topic.TRANSACTION_STATE_TOPIC_NAME + case _ => + throw new IllegalStateException(s"Unknown coordinator type $coordinatorType") + } + + val findCoordinatorRequest = new FindCoordinatorRequest.Builder( + new FindCoordinatorRequestData() + .setKeyType(coordinatorType.id()) + .setKey(groupId) + ).build(requestHeader.apiVersion) + val request = buildRequest(findCoordinatorRequest) + + val capturedResponse = expectNoThrottling() + + verifyTopicCreation(topicName, true, true, request) + + EasyMock.replay(replicaManager, clientRequestQuotaManager, requestChannel, authorizer, + autoTopicCreationManager, forwardingManager, controller, clientControllerQuotaManager, groupCoordinator, txnCoordinator) + + createKafkaApis(authorizer = Some(authorizer), + overrideProperties = topicConfigOverride).handleFindCoordinatorRequest(request) + + val response = readResponse(findCoordinatorRequest, capturedResponse) + .asInstanceOf[FindCoordinatorResponse] + assertEquals(Errors.COORDINATOR_NOT_AVAILABLE, response.error()) + + verify(authorizer, autoTopicCreationManager) + } + + @Test + def testMetadataAutoTopicCreationForOffsetTopic(): Unit = { + testMetadataAutoTopicCreation(Topic.GROUP_METADATA_TOPIC_NAME, enableAutoTopicCreation = true, + expectedError = Errors.UNKNOWN_TOPIC_OR_PARTITION) + } + + @Test + def testMetadataAutoTopicCreationForTxnTopic(): Unit = { + testMetadataAutoTopicCreation(Topic.TRANSACTION_STATE_TOPIC_NAME, enableAutoTopicCreation = true, + expectedError = Errors.UNKNOWN_TOPIC_OR_PARTITION) + } + + @Test + def testMetadataAutoTopicCreationForNonInternalTopic(): Unit = { + testMetadataAutoTopicCreation("topic", enableAutoTopicCreation = true, + expectedError = Errors.UNKNOWN_TOPIC_OR_PARTITION) + } + + @Test + def testMetadataAutoTopicCreationDisabledForOffsetTopic(): Unit = { + testMetadataAutoTopicCreation(Topic.GROUP_METADATA_TOPIC_NAME, enableAutoTopicCreation = false, + expectedError = Errors.UNKNOWN_TOPIC_OR_PARTITION) + } + + @Test + def testMetadataAutoTopicCreationDisabledForTxnTopic(): Unit = { + testMetadataAutoTopicCreation(Topic.TRANSACTION_STATE_TOPIC_NAME, enableAutoTopicCreation = false, + expectedError = Errors.UNKNOWN_TOPIC_OR_PARTITION) + } + + @Test + def testMetadataAutoTopicCreationDisabledForNonInternalTopic(): Unit = { + testMetadataAutoTopicCreation("topic", enableAutoTopicCreation = false, + expectedError = Errors.UNKNOWN_TOPIC_OR_PARTITION) + } + + @Test + def testMetadataAutoCreationDisabledForNonInternal(): Unit = { + testMetadataAutoTopicCreation("topic", enableAutoTopicCreation = true, + expectedError = Errors.UNKNOWN_TOPIC_OR_PARTITION) + } + + private def testMetadataAutoTopicCreation(topicName: String, + enableAutoTopicCreation: Boolean, + expectedError: Errors): Unit = { + val authorizer: Authorizer = EasyMock.niceMock(classOf[Authorizer]) + + val requestHeader = new RequestHeader(ApiKeys.METADATA, ApiKeys.METADATA.latestVersion, + clientId, 0) + + val numBrokersNeeded = 3 + addTopicToMetadataCache("some-topic", 1, 3) + + authorizeResource(authorizer, AclOperation.DESCRIBE, ResourceType.TOPIC, + topicName, AuthorizationResult.ALLOWED) + + if (enableAutoTopicCreation) + authorizeResource(authorizer, AclOperation.CREATE, ResourceType.CLUSTER, + Resource.CLUSTER_NAME, AuthorizationResult.ALLOWED, logIfDenied = false) + + val topicConfigOverride = mutable.Map.empty[String, String] + val isInternal = + topicName match { + case Topic.GROUP_METADATA_TOPIC_NAME => + topicConfigOverride.put(KafkaConfig.OffsetsTopicPartitionsProp, numBrokersNeeded.toString) + topicConfigOverride.put(KafkaConfig.OffsetsTopicReplicationFactorProp, numBrokersNeeded.toString) + EasyMock.expect(groupCoordinator.offsetsTopicConfigs).andReturn(new Properties) + true + + case Topic.TRANSACTION_STATE_TOPIC_NAME => + topicConfigOverride.put(KafkaConfig.TransactionsTopicPartitionsProp, numBrokersNeeded.toString) + topicConfigOverride.put(KafkaConfig.TransactionsTopicReplicationFactorProp, numBrokersNeeded.toString) + EasyMock.expect(txnCoordinator.transactionTopicConfigs).andReturn(new Properties) + true + case _ => + topicConfigOverride.put(KafkaConfig.NumPartitionsProp, numBrokersNeeded.toString) + topicConfigOverride.put(KafkaConfig.DefaultReplicationFactorProp, numBrokersNeeded.toString) + false + } + + val metadataRequest = new MetadataRequest.Builder( + List(topicName).asJava, enableAutoTopicCreation + ).build(requestHeader.apiVersion) + val request = buildRequest(metadataRequest) + + val capturedResponse = expectNoThrottling() + + verifyTopicCreation(topicName, enableAutoTopicCreation, isInternal, request) + + EasyMock.replay(replicaManager, clientRequestQuotaManager, requestChannel, authorizer, + autoTopicCreationManager, forwardingManager, clientControllerQuotaManager, groupCoordinator, txnCoordinator) + + createKafkaApis(authorizer = Some(authorizer), enableForwarding = enableAutoTopicCreation, + overrideProperties = topicConfigOverride).handleTopicMetadataRequest(request) + + val response = readResponse(metadataRequest, capturedResponse) + .asInstanceOf[MetadataResponse] + + val expectedMetadataResponse = util.Collections.singletonList(new TopicMetadata( + expectedError, + topicName, + isInternal, + util.Collections.emptyList() + )) + + assertEquals(expectedMetadataResponse, response.topicMetadata()) + + verify(authorizer, autoTopicCreationManager) + } + + private def verifyTopicCreation(topicName: String, + enableAutoTopicCreation: Boolean, + isInternal: Boolean, + request: RequestChannel.Request) = { + if (enableAutoTopicCreation) { + EasyMock.expect(clientControllerQuotaManager.newPermissiveQuotaFor(EasyMock.eq(request))) + .andReturn(UnboundedControllerMutationQuota) + + EasyMock.expect(autoTopicCreationManager.createTopics( + EasyMock.eq(Set(topicName)), + EasyMock.eq(UnboundedControllerMutationQuota))).andReturn( + Seq(new MetadataResponseTopic() + .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code()) + .setIsInternal(isInternal) + .setName(topicName)) + ).once() + } + } + + private def setupBrokerMetadata(hasEnoughLiveBrokers: Boolean, numBrokersNeeded: Int): Unit = { + addTopicToMetadataCache("some-topic", 1, + if (hasEnoughLiveBrokers) + numBrokersNeeded + else + numBrokersNeeded - 1) + } + @Test def testOffsetCommitWithInvalidPartition(): Unit = { val topic = "topic" @@ -1905,7 +2129,8 @@ class KafkaApisTest { val response = sendMetadataRequestWithInconsistentListeners(requestListener) assertFalse(createTopicIsCalled) - assertEquals(List("remaining-topic"), response.topicMetadata().asScala.map { metadata => metadata.topic() }) + val responseTopics = response.topicMetadata().asScala.map { metadata => metadata.topic() } + assertEquals(List("remaining-topic"), responseTopics) assertTrue(response.topicsByError(Errors.UNKNOWN_TOPIC_OR_PARTITION).isEmpty) } @@ -2602,7 +2827,7 @@ class KafkaApisTest { } def testUpdateMetadataRequest(currentBrokerEpoch: Long, brokerEpochInRequest: Long, expectedError: Errors): Unit = { - val updateMetadataRequest = createBasicMetadataRequest("topicA", 1, brokerEpochInRequest) + val updateMetadataRequest = createBasicMetadataRequest("topicA", 1, brokerEpochInRequest, 1) val request = buildRequest(updateMetadataRequest) val capturedResponse: Capture[RequestChannel.Response] = EasyMock.newCapture() @@ -3011,7 +3236,10 @@ class KafkaApisTest { capturedResponse } - private def createBasicMetadataRequest(topic: String, numPartitions: Int, brokerEpoch: Long): UpdateMetadataRequest = { + private def createBasicMetadataRequest(topic: String, + numPartitions: Int, + brokerEpoch: Long, + numBrokers: Int): UpdateMetadataRequest = { val replicas = List(0.asInstanceOf[Integer]).asJava def createPartitionState(partition: Int) = new UpdateMetadataPartitionState() @@ -3025,24 +3253,30 @@ class KafkaApisTest { .setIsr(replicas) val plaintextListener = ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT) - val broker = new UpdateMetadataBroker() - .setId(0) - .setRack("rack") - .setEndpoints(Seq(new UpdateMetadataEndpoint() - .setHost("broker0") - .setPort(9092) - .setSecurityProtocol(SecurityProtocol.PLAINTEXT.id) - .setListener(plaintextListener.value)).asJava) val partitionStates = (0 until numPartitions).map(createPartitionState) + val liveBrokers = (0 until numBrokers).map( + brokerId => createMetadataBroker(brokerId, plaintextListener)) new UpdateMetadataRequest.Builder(ApiKeys.UPDATE_METADATA.latestVersion, 0, - 0, brokerEpoch, partitionStates.asJava, Seq(broker).asJava, Collections.emptyMap()).build() + 0, brokerEpoch, partitionStates.asJava, liveBrokers.asJava, Collections.emptyMap()).build() } - private def addTopicToMetadataCache(topic: String, numPartitions: Int): Unit = { - val updateMetadataRequest = createBasicMetadataRequest(topic, numPartitions, 0) + private def addTopicToMetadataCache(topic: String, numPartitions: Int, numBrokers: Int = 1): Unit = { + val updateMetadataRequest = createBasicMetadataRequest(topic, numPartitions, 0, numBrokers) metadataCache.updateMetadata(correlationId = 0, updateMetadataRequest) } + private def createMetadataBroker(brokerId: Int, + listener: ListenerName): UpdateMetadataBroker = { + new UpdateMetadataBroker() + .setId(brokerId) + .setRack("rack") + .setEndpoints(Seq(new UpdateMetadataEndpoint() + .setHost("broker" + brokerId) + .setPort(9092) + .setSecurityProtocol(SecurityProtocol.PLAINTEXT.id) + .setListener(listener.value)).asJava) + } + @Test def testAlterReplicaLogDirs(): Unit = { val data = new AlterReplicaLogDirsRequestData() diff --git a/core/src/test/scala/unit/kafka/server/MetadataRequestTest.scala b/core/src/test/scala/unit/kafka/server/MetadataRequestTest.scala index 7b7078288efeb..01daf02245da5 100644 --- a/core/src/test/scala/unit/kafka/server/MetadataRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/MetadataRequestTest.scala @@ -17,14 +17,12 @@ package kafka.server -import java.util.{Optional, Properties} +import java.util.Optional -import kafka.network.SocketServer import kafka.utils.TestUtils import org.apache.kafka.common.Uuid import org.apache.kafka.common.errors.UnsupportedVersionException import org.apache.kafka.common.internals.Topic -import org.apache.kafka.common.message.MetadataRequestData import org.apache.kafka.common.protocol.Errors import org.apache.kafka.common.requests.{MetadataRequest, MetadataResponse} import org.apache.kafka.metadata.BrokerState @@ -32,16 +30,10 @@ import org.apache.kafka.test.TestUtils.isValidClusterId import org.junit.jupiter.api.Assertions._ import org.junit.jupiter.api.{BeforeEach, Test} -import scala.jdk.CollectionConverters._ import scala.collection.Seq +import scala.jdk.CollectionConverters._ -class MetadataRequestTest extends BaseRequestTest { - - override def brokerPropertyOverrides(properties: Properties): Unit = { - properties.setProperty(KafkaConfig.OffsetsTopicPartitionsProp, "1") - properties.setProperty(KafkaConfig.DefaultReplicationFactorProp, "2") - properties.setProperty(KafkaConfig.RackProp, s"rack/${properties.getProperty(KafkaConfig.BrokerIdProp)}") - } +class MetadataRequestTest extends AbstractMetadataRequestTest { @BeforeEach override def setUp(): Unit = { @@ -128,19 +120,12 @@ class MetadataRequestTest extends BaseRequestTest { @Test def testAutoTopicCreation(): Unit = { - def checkAutoCreatedTopic(autoCreatedTopic: String, response: MetadataResponse): Unit = { - assertEquals(Errors.LEADER_NOT_AVAILABLE, response.errors.get(autoCreatedTopic)) - assertEquals(Some(servers.head.config.numPartitions), zkClient.getTopicPartitionCount(autoCreatedTopic)) - for (i <- 0 until servers.head.config.numPartitions) - TestUtils.waitForPartitionMetadata(servers, autoCreatedTopic, i) - } - val topic1 = "t1" val topic2 = "t2" val topic3 = "t3" val topic4 = "t4" val topic5 = "t5" - createTopic(topic1, numPartitions = 1, replicationFactor = 1) + createTopic(topic1) val response1 = sendMetadataRequest(new MetadataRequest.Builder(Seq(topic1, topic2).asJava, true).build()) assertNull(response1.errors.get(topic1)) @@ -178,23 +163,24 @@ class MetadataRequestTest extends BaseRequestTest { @Test def testAutoCreateOfCollidingTopics(): Unit = { - val topic1 = "testAutoCreate_Topic" - val topic2 = "testAutoCreate.Topic" + val topic1 = "testAutoCreate.Topic" + val topic2 = "testAutoCreate_Topic" val response1 = sendMetadataRequest(new MetadataRequest.Builder(Seq(topic1, topic2).asJava, true).build) assertEquals(2, response1.topicMetadata.size) - var topicMetadata1 = response1.topicMetadata.asScala.head - val topicMetadata2 = response1.topicMetadata.asScala.toSeq(1) - assertEquals(Errors.LEADER_NOT_AVAILABLE, topicMetadata1.error) - assertEquals(topic1, topicMetadata1.topic) - assertEquals(Errors.INVALID_TOPIC_EXCEPTION, topicMetadata2.error) - assertEquals(topic2, topicMetadata2.topic) - TestUtils.waitUntilLeaderIsElectedOrChanged(zkClient, topic1, 0) - TestUtils.waitForPartitionMetadata(servers, topic1, 0) + val responseMap = response1.topicMetadata.asScala.map(metadata => (metadata.topic(), metadata.error)).toMap + + assertEquals(Set(topic1, topic2), responseMap.keySet) + // The topic creation will be delayed, and the name collision error will be swallowed. + assertEquals(Set(Errors.LEADER_NOT_AVAILABLE, Errors.INVALID_TOPIC_EXCEPTION), responseMap.values.toSet) + + val topicCreated = responseMap.head._1 + TestUtils.waitUntilLeaderIsElectedOrChanged(zkClient, topicCreated, 0) + TestUtils.waitForPartitionMetadata(servers, topicCreated, 0) // retry the metadata for the first auto created topic - val response2 = sendMetadataRequest(new MetadataRequest.Builder(Seq(topic1).asJava, true).build) - topicMetadata1 = response2.topicMetadata.asScala.head + val response2 = sendMetadataRequest(new MetadataRequest.Builder(Seq(topicCreated).asJava, true).build) + val topicMetadata1 = response2.topicMetadata.asScala.head assertEquals(Errors.NONE, topicMetadata1.error) assertEquals(Seq(Errors.NONE), topicMetadata1.partitionMetadata.asScala.map(_.error)) assertEquals(1, topicMetadata1.partitionMetadata.size) @@ -275,15 +261,6 @@ class MetadataRequestTest extends BaseRequestTest { } } - def requestData(topics: List[String], allowAutoTopicCreation: Boolean): MetadataRequestData = { - val data = new MetadataRequestData - if (topics == null) data.setTopics(null) - else topics.foreach(topic => data.topics.add(new MetadataRequestData.MetadataRequestTopic().setName(topic))) - - data.setAllowAutoTopicCreation(allowAutoTopicCreation) - data - } - @Test def testReplicaDownResponse(): Unit = { val replicaDownTopic = "replicaDown" @@ -397,9 +374,4 @@ class MetadataRequestTest extends BaseRequestTest { serverToShutdown.startup() checkMetadata(servers, servers.size) } - - private def sendMetadataRequest(request: MetadataRequest, destination: Option[SocketServer] = None): MetadataResponse = { - connectAndReceive[MetadataResponse](request, destination = destination.getOrElse(anySocketServer)) - } - } diff --git a/core/src/test/scala/unit/kafka/server/MetadataRequestWithForwardingTest.scala b/core/src/test/scala/unit/kafka/server/MetadataRequestWithForwardingTest.scala new file mode 100644 index 0000000000000..8409c968f0bc4 --- /dev/null +++ b/core/src/test/scala/unit/kafka/server/MetadataRequestWithForwardingTest.scala @@ -0,0 +1,111 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package kafka.server + +import kafka.utils.TestUtils +import org.apache.kafka.common.errors.UnsupportedVersionException +import org.apache.kafka.common.protocol.Errors +import org.apache.kafka.common.requests.MetadataRequest +import org.junit.jupiter.api.Assertions.{assertEquals, assertNull, assertThrows, assertTrue} +import org.junit.jupiter.api.{BeforeEach, Test} + +import scala.collection.Seq +import scala.jdk.CollectionConverters._ + +class MetadataRequestWithForwardingTest extends AbstractMetadataRequestTest { + + @BeforeEach + override def setUp(): Unit = { + doSetup(createOffsetsTopic = false) + } + + override def enableForwarding: Boolean = true + + @Test + def testAutoTopicCreation(): Unit = { + val topic1 = "t1" + val topic2 = "t2" + val topic3 = "t3" + val topic4 = "t4" + val topic5 = "t5" + createTopic(topic1) + + val response1 = sendMetadataRequest(new MetadataRequest.Builder(Seq(topic1, topic2).asJava, true).build()) + assertNull(response1.errors.get(topic1)) + checkAutoCreatedTopic(topic2, response1) + + // The default behavior in old versions of the metadata API is to allow topic creation, so + // protocol downgrades should happen gracefully when auto-creation is explicitly requested. + val response2 = sendMetadataRequest(new MetadataRequest.Builder(Seq(topic3).asJava, true).build(1)) + checkAutoCreatedTopic(topic3, response2) + + // V3 doesn't support a configurable allowAutoTopicCreation, so disabling auto-creation is not supported + assertThrows(classOf[UnsupportedVersionException], () => sendMetadataRequest(new MetadataRequest(requestData(List(topic4), false), 3.toShort))) + + // V4 and higher support a configurable allowAutoTopicCreation + val response3 = sendMetadataRequest(new MetadataRequest.Builder(Seq(topic4, topic5).asJava, false, 4.toShort).build) + assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION, response3.errors.get(topic4)) + assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION, response3.errors.get(topic5)) + assertEquals(None, zkClient.getTopicPartitionCount(topic5)) + } + + @Test + def testAutoCreateTopicWithInvalidReplicationFactor(): Unit = { + // Shutdown all but one broker so that the number of brokers is less than the default replication factor + servers.tail.foreach(_.shutdown()) + servers.tail.foreach(_.awaitShutdown()) + + val topic1 = "testAutoCreateTopic" + val response1 = sendMetadataRequest(new MetadataRequest.Builder(Seq(topic1).asJava, true).build) + assertEquals(1, response1.topicMetadata.size) + val topicMetadata = response1.topicMetadata.asScala.head + assertEquals(Errors.INVALID_REPLICATION_FACTOR, topicMetadata.error) + assertEquals(topic1, topicMetadata.topic) + assertEquals(0, topicMetadata.partitionMetadata.size) + } + + @Test + def testAutoCreateOfCollidingTopics(): Unit = { + val topic1 = "testAutoCreate.Topic" + val topic2 = "testAutoCreate_Topic" + val response1 = sendMetadataRequest(new MetadataRequest.Builder(Seq(topic1, topic2).asJava, true).build) + assertEquals(2, response1.topicMetadata.size) + + val responseMap = response1.topicMetadata.asScala.map(metadata => (metadata.topic(), metadata.error)).toMap + + assertEquals(Set(topic1, topic2), responseMap.keySet) + // The topic creation will be delayed, and the name collision error will be swallowed. + assertEquals(Set(Errors.LEADER_NOT_AVAILABLE, Errors.INVALID_TOPIC_EXCEPTION), responseMap.values.toSet) + + val topicCreated = responseMap.head._1 + TestUtils.waitUntilLeaderIsElectedOrChanged(zkClient, topicCreated, 0) + TestUtils.waitForPartitionMetadata(servers, topicCreated, 0) + + // retry the metadata for the first auto created topic + val response2 = sendMetadataRequest(new MetadataRequest.Builder(Seq(topicCreated).asJava, true).build) + val topicMetadata1 = response2.topicMetadata.asScala.head + assertEquals(Errors.NONE, topicMetadata1.error) + assertEquals(Seq(Errors.NONE), topicMetadata1.partitionMetadata.asScala.map(_.error)) + assertEquals(1, topicMetadata1.partitionMetadata.size) + val partitionMetadata = topicMetadata1.partitionMetadata.asScala.head + assertEquals(0, partitionMetadata.partition) + assertEquals(2, partitionMetadata.replicaIds.size) + assertTrue(partitionMetadata.leaderId.isPresent) + assertTrue(partitionMetadata.leaderId.get >= 0) + } +} diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/MetadataRequestBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/MetadataRequestBenchmark.java index cdbf9be694617..887d53dad0b0b 100644 --- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/MetadataRequestBenchmark.java +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/MetadataRequestBenchmark.java @@ -22,6 +22,7 @@ import kafka.coordinator.transaction.TransactionCoordinator; import kafka.network.RequestChannel; import kafka.network.RequestConvertToJson; +import kafka.server.AutoTopicCreationManager; import kafka.server.BrokerFeatures; import kafka.server.BrokerTopicStats; import kafka.server.ClientQuotaManager; @@ -100,6 +101,7 @@ public class MetadataRequestBenchmark { private ZkAdminManager adminManager = Mockito.mock(ZkAdminManager.class); private TransactionCoordinator transactionCoordinator = Mockito.mock(TransactionCoordinator.class); private KafkaController kafkaController = Mockito.mock(KafkaController.class); + private AutoTopicCreationManager autoTopicCreationManager = Mockito.mock(AutoTopicCreationManager.class); private KafkaZkClient kafkaZkClient = Mockito.mock(KafkaZkClient.class); private Metrics metrics = new Metrics(); private int brokerId = 1; @@ -175,6 +177,7 @@ private KafkaApis createKafkaApis() { replicaManager, groupCoordinator, transactionCoordinator, + autoTopicCreationManager, brokerId, new KafkaConfig(kafkaProps), new CachedConfigRepository(), diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java index 9f18bcaa21ffd..d866954668618 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java @@ -44,13 +44,13 @@ public class StandbyTask extends AbstractTask implements Task { private final StreamsMetricsImpl streamsMetrics; /** - * @param id the ID of this task - * @param inputPartitions input topic partitions, used for thread metadata only - * @param topology the instance of {@link ProcessorTopology} - * @param config the {@link StreamsConfig} specified by the user - * @param streamsMetrics the {@link StreamsMetrics} created by the thread - * @param stateMgr the {@link ProcessorStateManager} for this task - * @param stateDirectory the {@link StateDirectory} created by the thread + * @param id the ID of this task + * @param partitions input topic partitions, used for thread metadata only + * @param topology the instance of {@link ProcessorTopology} + * @param config the {@link StreamsConfig} specified by the user + * @param streamsMetrics the {@link StreamsMetrics} created by the thread + * @param stateMgr the {@link ProcessorStateManager} for this task + * @param stateDirectory the {@link StateDirectory} created by the thread */ StandbyTask(final TaskId id, final Set inputPartitions, @@ -111,7 +111,7 @@ public void initializeIfNeeded() { } @Override - public void completeRestoration(final java.util.function.Consumer> offsetResetter) { + public void completeRestoration() { throw new IllegalStateException("Standby task " + id + " should never be completing restoration"); } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java index fa3d2d9442566..fbf010765fe0d 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java @@ -85,7 +85,6 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator, private final RecordCollector recordCollector; private final PartitionGroup.RecordInfo recordInfo; private final Map consumedOffsets; - private final Set resetOffsetsForPartitions; private final PunctuationQueue streamTimePunctuationQueue; private final PunctuationQueue systemTimePunctuationQueue; private final StreamsMetricsImpl streamsMetrics; @@ -168,7 +167,6 @@ public StreamTask(final TaskId id, // initialize the consumed and committed offset cache consumedOffsets = new HashMap<>(); - resetOffsetsForPartitions = new HashSet<>(); recordQueueCreator = new RecordQueueCreator(this.logContext, config.defaultTimestampExtractor(), config.defaultDeserializationExceptionHandler()); @@ -231,22 +229,17 @@ public void initializeIfNeeded() { } } - public void addPartitionsForOffsetReset(final Set partitionsForOffsetReset) { - mainConsumer.pause(partitionsForOffsetReset); - resetOffsetsForPartitions.addAll(partitionsForOffsetReset); - } - /** * @throws TimeoutException if fetching committed offsets timed out */ @Override - public void completeRestoration(final java.util.function.Consumer> offsetResetter) { + public void completeRestoration() { switch (state()) { case RUNNING: return; case RESTORING: - resetOffsetsIfNeededAndInitializeMetadata(offsetResetter); + initializeMetadata(); initializeTopology(); processorContext.initialize(); @@ -839,27 +832,12 @@ private Map checkpointableOffsets() { return checkpointableOffsets; } - private void resetOffsetsIfNeededAndInitializeMetadata(final java.util.function.Consumer> offsetResetter) { + private void initializeMetadata() { try { - final Map offsetsAndMetadata = mainConsumer.committed(inputPartitions()); - - for (final Map.Entry committedEntry : offsetsAndMetadata.entrySet()) { - if (resetOffsetsForPartitions.contains(committedEntry.getKey())) { - final OffsetAndMetadata offsetAndMetadata = committedEntry.getValue(); - if (offsetAndMetadata != null) { - mainConsumer.seek(committedEntry.getKey(), offsetAndMetadata); - resetOffsetsForPartitions.remove(committedEntry.getKey()); - } - } - } - - offsetResetter.accept(resetOffsetsForPartitions); - resetOffsetsForPartitions.clear(); - - initializeTaskTime(offsetsAndMetadata.entrySet().stream() - .filter(e -> e.getValue() != null) - .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)) - ); + final Map offsetsAndMetadata = mainConsumer.committed(inputPartitions()).entrySet().stream() + .filter(e -> e.getValue() != null) + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + initializeTaskTime(offsetsAndMetadata); } catch (final TimeoutException timeoutException) { log.warn( "Encountered {} while trying to fetch committed offsets, will retry initializing the metadata in the next loop." + diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java index fa886b35e411f..47eb965724859 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java @@ -421,6 +421,8 @@ public static StreamThread create(final InternalTopologyBuilder builder, cache::resize ); + taskManager.setPartitionResetter(partitions -> streamThread.resetOffsets(partitions, null)); + return streamThread.updateThreadMetadata(getSharedAdminClientId(clientId)); } @@ -837,7 +839,7 @@ private void initializeAndRestorePhase() { // transit to restore active is idempotent so we can call it multiple times changelogReader.enforceRestoreActive(); - if (taskManager.tryToCompleteRestoration(now, partitions -> resetOffsets(partitions, null))) { + if (taskManager.tryToCompleteRestoration(now)) { changelogReader.transitToUpdateStandby(); log.info("Restoration took {} ms for all tasks {}", time.milliseconds() - lastPartitionAssignedMs, taskManager.tasks().keySet()); diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java index af8af4d19e5f2..8e3aa0b41e854 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java @@ -109,14 +109,10 @@ enum TaskType { */ void initializeIfNeeded(); - default void addPartitionsForOffsetReset(final Set partitionsForOffsetReset) { - throw new UnsupportedOperationException(); - } - /** * @throws StreamsException fatal error, should close the thread */ - void completeRestoration(final java.util.function.Consumer> offsetResetter); + void completeRestoration(); void suspend(); diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java index 430f0c6ee45c5..19e91e2be5372 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java @@ -90,6 +90,7 @@ public class TaskManager { // includes assigned & initialized tasks and unassigned tasks we locked temporarily during rebalance private final Set lockedTaskDirectories = new HashSet<>(); + private java.util.function.Consumer> resetter; TaskManager(final Time time, final ChangelogReader changelogReader, @@ -226,7 +227,18 @@ private void closeAndRevive(final Map> taskWith ); } - task.addPartitionsForOffsetReset(assignedToPauseAndReset); + mainConsumer.pause(assignedToPauseAndReset); + // TODO: KIP-572 need to handle `TimeoutException` + final Map committed = mainConsumer.committed(assignedToPauseAndReset); + for (final Map.Entry committedEntry : committed.entrySet()) { + final OffsetAndMetadata offsetAndMetadata = committedEntry.getValue(); + if (offsetAndMetadata != null) { + mainConsumer.seek(committedEntry.getKey(), offsetAndMetadata); + assignedToPauseAndReset.remove(committedEntry.getKey()); + } + } + // throws if anything has no configured reset policy + resetter.accept(assignedToPauseAndReset); } task.revive(); } @@ -407,7 +419,7 @@ private void handleCloseAndRecycle(final Set tasksToRecycle, * @throws StreamsException if the store's change log does not contain the partition * @return {@code true} if all tasks are fully restored */ - boolean tryToCompleteRestoration(final long now, final java.util.function.Consumer> offsetResetter) { + boolean tryToCompleteRestoration(final long now) { boolean allRunning = true; final List activeTasks = new LinkedList<>(); @@ -441,7 +453,7 @@ boolean tryToCompleteRestoration(final long now, final java.util.function.Consum for (final Task task : activeTasks) { if (restored.containsAll(task.changelogPartitions())) { try { - task.completeRestoration(offsetResetter); + task.completeRestoration(); task.clearTaskTimeout(); } catch (final TimeoutException timeoutException) { task.maybeInitTaskTimeoutOrThrow(now, timeoutException); @@ -1255,6 +1267,10 @@ boolean needsInitializationOrRestoration() { return tasks().values().stream().anyMatch(Task::needsInitializationOrRestoration); } + public void setPartitionResetter(final java.util.function.Consumer> resetter) { + this.resetter = resetter; + } + // for testing only void addTask(final Task task) { tasks.addTask(task); diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java index 5de6b7474de27..879ce0666520f 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java @@ -98,7 +98,7 @@ public class EosIntegrationTest { @ClassRule public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster( NUM_BROKERS, - Utils.mkProperties(Collections.singletonMap("auto.create.topics.enable", "false")) + Utils.mkProperties(Collections.singletonMap("auto.create.topics.enable", "true")) ); private String applicationId; diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java index 4a0e3ac1bea95..456697bbc3368 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java @@ -167,7 +167,7 @@ public void shouldThrowLockExceptionIfFailedToLockStateDirectory() throws IOExce task = createStandbyTask(); - assertThrows(LockException.class, () -> task.initializeIfNeeded()); + assertThrows(LockException.class, task::initializeIfNeeded); task = null; } diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java index 1469b87b850e0..ea3fbdd91e1cf 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java @@ -83,7 +83,6 @@ import java.util.List; import java.util.Map; import java.util.Set; -import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; import java.util.stream.Collectors; @@ -278,7 +277,7 @@ public void shouldThrowLockExceptionIfFailedToLockStateDirectory() throws IOExce task = createStatefulTask(createConfig("100"), false); - assertThrows(LockException.class, () -> task.initializeIfNeeded()); + assertThrows(LockException.class, task::initializeIfNeeded); } @Test @@ -328,69 +327,6 @@ public void shouldAttemptToDeleteStateDirectoryWhenCloseDirtyAndEosEnabled() thr ctrl.verify(); } - @Test - public void shouldResetOffsetsToLastCommittedForSpecifiedPartitions() { - task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); - task.addPartitionsForOffsetReset(Collections.singleton(partition1)); - - consumer.seek(partition1, 5L); - consumer.commitSync(); - - consumer.seek(partition1, 10L); - consumer.seek(partition2, 15L); - - final java.util.function.Consumer> resetter = - EasyMock.mock(java.util.function.Consumer.class); - resetter.accept(Collections.emptySet()); - EasyMock.expectLastCall(); - EasyMock.replay(resetter); - - task.initializeIfNeeded(); - task.completeRestoration(resetter); - - assertThat(consumer.position(partition1), equalTo(5L)); - assertThat(consumer.position(partition2), equalTo(15L)); - } - - @Test - public void shouldAutoOffsetResetIfNoCommittedOffsetFound() { - task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); - task.addPartitionsForOffsetReset(Collections.singleton(partition1)); - - final AtomicReference shouldNotSeek = new AtomicReference<>(); - final MockConsumer consumer = new MockConsumer(OffsetResetStrategy.EARLIEST) { - @Override - public void seek(final TopicPartition partition, final long offset) { - final AssertionError error = shouldNotSeek.get(); - if (error != null) { - throw error; - } - super.seek(partition, offset); - } - }; - consumer.assign(asList(partition1, partition2)); - consumer.updateBeginningOffsets(mkMap(mkEntry(partition1, 0L), mkEntry(partition2, 0L))); - - consumer.seek(partition1, 5L); - consumer.seek(partition2, 15L); - - shouldNotSeek.set(new AssertionError("Should not seek")); - - final java.util.function.Consumer> resetter = - EasyMock.mock(java.util.function.Consumer.class); - resetter.accept(Collections.singleton(partition1)); - EasyMock.expectLastCall(); - EasyMock.replay(resetter); - - task.initializeIfNeeded(); - task.completeRestoration(resetter); - - // because we mocked the `resetter` positions don't change - assertThat(consumer.position(partition1), equalTo(5L)); - assertThat(consumer.position(partition2), equalTo(15L)); - EasyMock.verify(resetter); - } - @Test public void shouldReadCommittedStreamTimeOnInitialize() { stateDirectory = EasyMock.createNiceMock(StateDirectory.class); @@ -404,7 +340,7 @@ public void shouldReadCommittedStreamTimeOnInitialize() { assertEquals(RecordQueue.UNKNOWN, task.streamTime()); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); assertEquals(10L, task.streamTime()); } @@ -435,7 +371,7 @@ public void shouldTransitToRestoringThenRunningAfterCreation() throws IOExceptio assertEquals(RESTORING, task.state()); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); assertEquals(RUNNING, task.state()); assertTrue(source1.initialized); @@ -644,7 +580,7 @@ public void process(final Record record) { task = createStatelessTaskWithForwardingTopology(evenKeyForwardingSourceNode); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); final String sourceNodeName = evenKeyForwardingSourceNode.name(); final String terminalNodeName = processorStreamTime.name(); @@ -960,7 +896,7 @@ public void shouldPauseAndResumeBasedOnBufferedRecords() { public void shouldPunctuateOnceStreamTimeAfterGap() { task = createStatelessTask(createConfig(), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.addRecords(partition1, asList( getConsumerRecordWithOffsetAsTimestamp(partition1, 20), @@ -1048,7 +984,7 @@ public void shouldPunctuateOnceStreamTimeAfterGap() { public void shouldRespectPunctuateCancellationStreamTime() { task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.addRecords(partition1, asList( getConsumerRecordWithOffsetAsTimestamp(partition1, 20), @@ -1088,7 +1024,7 @@ public void shouldRespectPunctuateCancellationStreamTime() { public void shouldRespectPunctuateCancellationSystemTime() { task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); final long now = time.milliseconds(); time.sleep(10); assertTrue(task.maybePunctuateSystemTime()); @@ -1102,7 +1038,7 @@ public void shouldRespectPunctuateCancellationSystemTime() { public void shouldRespectCommitNeeded() { task = createSingleSourceStateless(createConfig(AT_LEAST_ONCE, "0"), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); assertFalse(task.commitNeeded()); @@ -1140,7 +1076,7 @@ public void shouldRespectCommitNeeded() { public void shouldCommitNextOffsetFromQueueIfAvailable() { task = createSingleSourceStateless(createConfig(AT_LEAST_ONCE, "0"), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.addRecords(partition1, asList( getConsumerRecordWithOffsetAsTimestamp(partition1, 0L), @@ -1159,7 +1095,7 @@ public void shouldCommitNextOffsetFromQueueIfAvailable() { public void shouldCommitConsumerPositionIfRecordQueueIsEmpty() { task = createStatelessTask(createConfig(), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); consumer.addRecord(getConsumerRecordWithOffsetAsTimestamp(partition1, 0L)); consumer.addRecord(getConsumerRecordWithOffsetAsTimestamp(partition1, 1L)); @@ -1193,7 +1129,7 @@ public void shouldFailOnCommitIfTaskIsClosed() { public void shouldRespectCommitRequested() { task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.requestCommit(); assertTrue(task.commitRequested()); @@ -1225,7 +1161,7 @@ public void shouldReturnUnknownTimestampIfEmptyMessage() { public void shouldBeProcessableIfAllPartitionsBuffered() { task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); assertFalse(task.process(0L)); @@ -1243,7 +1179,7 @@ public void shouldBeProcessableIfAllPartitionsBuffered() { public void shouldPunctuateSystemTimeWhenIntervalElapsed() { task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); final long now = time.milliseconds(); time.sleep(10); assertTrue(task.maybePunctuateSystemTime()); @@ -1263,7 +1199,7 @@ public void shouldPunctuateSystemTimeWhenIntervalElapsed() { public void shouldNotPunctuateSystemTimeWhenIntervalNotElapsed() { task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); assertFalse(task.maybePunctuateSystemTime()); time.sleep(9); assertFalse(task.maybePunctuateSystemTime()); @@ -1274,7 +1210,7 @@ public void shouldNotPunctuateSystemTimeWhenIntervalNotElapsed() { public void shouldPunctuateOnceSystemTimeAfterGap() { task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); final long now = time.milliseconds(); time.sleep(100); assertTrue(task.maybePunctuateSystemTime()); @@ -1300,7 +1236,7 @@ public void shouldPunctuateOnceSystemTimeAfterGap() { public void shouldWrapKafkaExceptionsWithStreamsExceptionAndAddContextWhenPunctuatingStreamTime() { task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); try { task.punctuate(processorStreamTime, 1, PunctuationType.STREAM_TIME, timestamp -> { @@ -1318,7 +1254,7 @@ public void shouldWrapKafkaExceptionsWithStreamsExceptionAndAddContextWhenPunctu public void shouldWrapKafkaExceptionsWithStreamsExceptionAndAddContextWhenPunctuatingWallClockTimeTime() { task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); try { task.punctuate(processorSystemTime, 1, PunctuationType.WALL_CLOCK_TIME, timestamp -> { @@ -1336,7 +1272,7 @@ public void shouldWrapKafkaExceptionsWithStreamsExceptionAndAddContextWhenPunctu public void shouldNotShareHeadersBetweenPunctuateIterations() { task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.punctuate( processorSystemTime, @@ -1362,7 +1298,7 @@ public void shouldWrapKafkaExceptionWithStreamsExceptionWhenProcess() { task = createFaultyStatefulTask(createConfig("100")); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.addRecords(partition1, asList( getConsumerRecordWithOffsetAsTimestamp(partition1, 10), @@ -1389,9 +1325,9 @@ public void shouldReadCommittedOffsetAndRethrowTimeoutWhenCompleteRestoration() task = createDisconnectedTask(createConfig("100")); - task.transitionTo(RESTORING); + task.initializeIfNeeded(); - assertThrows(TimeoutException.class, () -> task.completeRestoration(noOpResetter -> { })); + assertThrows(TimeoutException.class, task::completeRestoration); } @Test @@ -1419,7 +1355,7 @@ public void shouldReInitializeTopologyWhenResuming() throws IOException { assertFalse(source1.initialized); assertFalse(source2.initialized); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); assertEquals(RUNNING, task.state()); assertTrue(source1.initialized); @@ -1448,7 +1384,7 @@ public void shouldNotCheckpointOffsetsAgainOnCommitIfSnapshotNotChangedMuch() { task = createStatefulTask(createConfig("100"), true); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.prepareCommit(); task.postCommit(true); // should checkpoint @@ -1478,7 +1414,7 @@ public void shouldCheckpointOffsetsOnCommitIfSnapshotMuchChanged() { task = createStatefulTask(createConfig("100"), true); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.prepareCommit(); task.postCommit(true); @@ -1499,7 +1435,7 @@ public void shouldNotCheckpointOffsetsOnCommitIfEosIsEnabled() { task = createStatefulTask(createConfig(StreamsConfig.EXACTLY_ONCE, "100"), true); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.prepareCommit(); task.postCommit(false); final File checkpointFile = new File( @@ -1514,7 +1450,7 @@ public void shouldNotCheckpointOffsetsOnCommitIfEosIsEnabled() { public void shouldThrowIllegalStateExceptionIfCurrentNodeIsNotNullWhenPunctuateCalled() { task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.processorContext().setCurrentNode(processorStreamTime); try { task.punctuate(processorStreamTime, 10, PunctuationType.STREAM_TIME, punctuator); @@ -1528,7 +1464,7 @@ public void shouldThrowIllegalStateExceptionIfCurrentNodeIsNotNullWhenPunctuateC public void shouldCallPunctuateOnPassedInProcessorNode() { task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.punctuate(processorStreamTime, 5, PunctuationType.STREAM_TIME, punctuator); assertThat(punctuatedAt, equalTo(5L)); task.punctuate(processorStreamTime, 10, PunctuationType.STREAM_TIME, punctuator); @@ -1539,7 +1475,7 @@ public void shouldCallPunctuateOnPassedInProcessorNode() { public void shouldSetProcessorNodeOnContextBackToNullAfterSuccessfulPunctuate() { task = createStatelessTask(createConfig("100"), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.punctuate(processorStreamTime, 5, PunctuationType.STREAM_TIME, punctuator); assertThat(task.processorContext().currentNode(), nullValue()); } @@ -1569,7 +1505,7 @@ public void shouldCloseStateManagerEvenDuringFailureOnUncleanTaskClose() { task = createFaultyStatefulTask(createConfig("100")); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); assertThrows(RuntimeException.class, () -> task.suspend()); task.closeDirty(); @@ -1618,7 +1554,7 @@ public void shouldReturnOffsetsForRepartitionTopicsForPurging() { logContext); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.addRecords(partition1, singletonList(getConsumerRecordWithOffsetAsTimestamp(partition1, 5L))); task.addRecords(repartition, singletonList(getConsumerRecordWithOffsetAsTimestamp(repartition, 10L))); @@ -1648,9 +1584,9 @@ public Map committed(final Set task.completeRestoration(noOpResetter -> { })); + assertThrows(StreamsException.class, task::completeRestoration); } @Test @@ -1708,7 +1644,7 @@ public void shouldNotCheckpointForSuspendedRunningTaskWithSmallProgress() { task = createStatefulTask(createConfig("100"), true); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.prepareCommit(); task.postCommit(false); @@ -1729,7 +1665,7 @@ public void shouldCheckpointForSuspendedRunningTaskWithLargeProgress() { task = createStatefulTask(createConfig("100"), true); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.prepareCommit(); task.postCommit(false); @@ -1753,7 +1689,7 @@ public void shouldCheckpointWhileUpdateSnapshotWithTheConsumedOffsetsForSuspende task = createStatefulTask(createConfig(), true); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.addRecords(partition1, singleton(getConsumerRecordWithOffsetAsTimestamp(partition1, 10))); task.addRecords(partition2, singleton(getConsumerRecordWithOffsetAsTimestamp(partition2, 10))); task.process(100L); @@ -1777,7 +1713,7 @@ public void shouldReturnStateManagerChangelogOffsets() { assertEquals(singletonMap(partition1, 50L), task.changelogOffsets()); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); assertEquals(singletonMap(partition1, Task.LATEST_OFFSET), task.changelogOffsets()); } @@ -1822,7 +1758,7 @@ public void shouldCheckpointOnCloseRestoringIfNoProgress() { task = createOptimizedStatefulTask(createConfig("100"), consumer); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.suspend(); task.prepareCommit(); task.postCommit(true); @@ -1849,7 +1785,7 @@ public void shouldCheckpointOffsetsOnPostCommit() { task = createOptimizedStatefulTask(createConfig(), consumer); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.addRecords(partition1, singletonList(getConsumerRecordWithOffsetAsTimestamp(partition1, consumedOffset))); task.process(100L); @@ -1880,7 +1816,7 @@ public void shouldThrowExceptionOnCloseCleanError() { task = createOptimizedStatefulTask(createConfig("100"), consumer); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.addRecords(partition1, singletonList(getConsumerRecordWithOffsetAsTimestamp(partition1, offset))); task.process(100L); @@ -1922,7 +1858,7 @@ public void shouldThrowOnCloseCleanFlushError() { task = createOptimizedStatefulTask(createConfig("100"), consumer); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); // process one record to make commit needed task.addRecords(partition1, singletonList(getConsumerRecordWithOffsetAsTimestamp(partition1, offset))); @@ -2074,7 +2010,7 @@ public void shouldUpdatePartitions() { public void shouldThrowIfCleanClosingDirtyTask() { task = createSingleSourceStateless(createConfig(AT_LEAST_ONCE, "0"), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.addRecords(partition1, singletonList(getConsumerRecordWithOffsetAsTimestamp(partition1, 0))); assertTrue(task.process(0L)); @@ -2087,7 +2023,7 @@ public void shouldThrowIfCleanClosingDirtyTask() { public void shouldThrowIfRecyclingDirtyTask() { task = createStatelessTask(createConfig(), StreamsConfig.METRICS_LATEST); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.addRecords(partition1, singletonList(getConsumerRecordWithOffsetAsTimestamp(partition1, 0))); task.addRecords(partition2, singletonList(getConsumerRecordWithOffsetAsTimestamp(partition2, 0))); @@ -2110,7 +2046,7 @@ public void shouldOnlyRecycleSuspendedTasks() { task.initializeIfNeeded(); assertThrows(IllegalStateException.class, () -> task.closeCleanAndRecycleState()); // RESTORING - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); assertThrows(IllegalStateException.class, () -> task.closeCleanAndRecycleState()); // RUNNING task.suspend(); @@ -2145,7 +2081,7 @@ public void shouldAlwaysSuspendRunningTasks() { EasyMock.replay(stateManager); task = createFaultyStatefulTask(createConfig("100")); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); assertThat(task.state(), equalTo(RUNNING)); assertThrows(RuntimeException.class, () -> task.suspend()); assertThat(task.state(), equalTo(SUSPENDED)); diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java index 95018715a69fb..2f95a208fa38d 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java @@ -1622,7 +1622,7 @@ public void shouldUpdateStandbyTask() throws Exception { standbyTasks.put(task3, Collections.singleton(t2p1)); thread.taskManager().handleAssignment(emptyMap(), standbyTasks); - thread.taskManager().tryToCompleteRestoration(mockTime.milliseconds(), null); + thread.taskManager().tryToCompleteRestoration(mockTime.milliseconds()); thread.rebalanceListener().onPartitionsAssigned(Collections.emptyList()); diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java index 9a3a8c92c3032..36224e0cd5c00 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java @@ -466,7 +466,7 @@ public void shouldCloseActiveUnassignedSuspendedTasksWhenClosingRevokedTasks() { replay(activeTaskCreator, standbyTaskCreator, topologyBuilder, consumer, changeLogReader); taskManager.handleAssignment(taskId00Assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); taskManager.handleRevocation(taskId00Partitions); @@ -545,7 +545,7 @@ public void shouldCloseActiveTasksWhenHandlingLostTasks() throws Exception { replay(activeTaskCreator, standbyTaskCreator, topologyBuilder, consumer, changeLogReader); taskManager.handleAssignment(taskId00Assignment, taskId01Assignment); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); assertThat(task01.state(), is(Task.State.RUNNING)); @@ -599,7 +599,7 @@ public void shouldThrowWhenHandlingClosingTasksOnProducerCloseError() { replay(activeTaskCreator, standbyTaskCreator, topologyBuilder, consumer, changeLogReader); taskManager.handleAssignment(taskId00Assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); taskManager.handleRevocation(taskId00Partitions); @@ -640,17 +640,24 @@ public void postCommit(final boolean enforceCheckpoint) { expect(activeTaskCreator.createTasks(anyObject(), eq(taskId00Assignment))).andStubReturn(singletonList(task00)); topologyBuilder.addSubscribedTopicsFromAssignment(anyObject(), anyString()); expectLastCall().anyTimes(); + expect(consumer.assignment()).andReturn(taskId00Partitions); + consumer.pause(taskId00Partitions); + expectLastCall(); + final OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(0L); + expect(consumer.committed(taskId00Partitions)).andReturn(singletonMap(t1p0, offsetAndMetadata)); + consumer.seek(t1p0, offsetAndMetadata); + expectLastCall(); replay(activeTaskCreator, topologyBuilder, consumer, changeLogReader); - + taskManager.setPartitionResetter(tp -> assertThat(tp, is(empty()))); taskManager.handleAssignment(taskId00Assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), tp -> assertThat(tp, is(empty()))), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); taskManager.handleCorruption(singletonMap(taskId00, taskId00Partitions)); + assertThat(task00.commitPrepared, is(true)); assertThat(task00.state(), is(Task.State.CREATED)); - assertThat(task00.partitionsForOffsetReset, equalTo(taskId00Partitions)); assertThat(enforcedCheckpoint.get(), is(true)); assertThat(taskManager.activeTaskMap(), is(singletonMap(taskId00, task00))); assertThat(taskManager.standbyTaskMap(), Matchers.anEmptyMap()); @@ -678,16 +685,22 @@ public void suspend() { topologyBuilder.addSubscribedTopicsFromAssignment(anyObject(), anyString()); expectLastCall().anyTimes(); expect(consumer.assignment()).andReturn(taskId00Partitions); + consumer.pause(taskId00Partitions); + expectLastCall(); + final OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(0L); + expect(consumer.committed(taskId00Partitions)).andReturn(singletonMap(t1p0, offsetAndMetadata)); + consumer.seek(t1p0, offsetAndMetadata); + expectLastCall(); + taskManager.setPartitionResetter(tp -> assertThat(tp, is(empty()))); replay(activeTaskCreator, topologyBuilder, consumer, changeLogReader); taskManager.handleAssignment(taskId00Assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), tp -> assertThat(tp, is(empty()))), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); taskManager.handleCorruption(singletonMap(taskId00, taskId00Partitions)); assertThat(task00.commitPrepared, is(true)); assertThat(task00.state(), is(Task.State.CREATED)); - assertThat(task00.partitionsForOffsetReset, equalTo(taskId00Partitions)); assertThat(taskManager.activeTaskMap(), is(singletonMap(taskId00, task00))); assertThat(taskManager.standbyTaskMap(), Matchers.anEmptyMap()); @@ -712,22 +725,27 @@ public void shouldCommitNonCorruptedTasksOnTaskCorruptedException() { .andStubReturn(asList(corruptedTask, nonCorruptedTask)); topologyBuilder.addSubscribedTopicsFromAssignment(anyObject(), anyString()); expectLastCall().anyTimes(); + expectRestoreToBeCompleted(consumer, changeLogReader); consumer.commitSync(eq(emptyMap())); expect(consumer.assignment()).andReturn(taskId00Partitions); + consumer.pause(taskId00Partitions); + expectLastCall(); + final OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(0L); + expect(consumer.committed(taskId00Partitions)).andReturn(singletonMap(t1p0, offsetAndMetadata)); + consumer.seek(t1p0, offsetAndMetadata); + expectLastCall(); replay(activeTaskCreator, topologyBuilder, consumer, changeLogReader); - + taskManager.setPartitionResetter(tp -> assertThat(tp, is(empty()))); taskManager.handleAssignment(assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), tp -> assertThat(tp, is(empty()))), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(nonCorruptedTask.state(), is(Task.State.RUNNING)); nonCorruptedTask.setCommitNeeded(); taskManager.handleCorruption(singletonMap(taskId00, taskId00Partitions)); - assertTrue(nonCorruptedTask.commitPrepared); - assertThat(nonCorruptedTask.partitionsForOffsetReset, equalTo(Collections.emptySet())); - assertThat(corruptedTask.partitionsForOffsetReset, equalTo(taskId00Partitions)); + assertTrue(nonCorruptedTask.commitPrepared); verify(consumer); } @@ -750,15 +768,21 @@ public void shouldNotCommitNonRunningNonCorruptedTasks() { .andStubReturn(asList(corruptedTask, nonRunningNonCorruptedTask)); topologyBuilder.addSubscribedTopicsFromAssignment(anyObject(), anyString()); expectLastCall().anyTimes(); + expect(consumer.assignment()).andReturn(taskId00Partitions); - replay(activeTaskCreator, topologyBuilder, consumer, changeLogReader); + consumer.pause(taskId00Partitions); + expectLastCall(); + final OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(0L); + expect(consumer.committed(taskId00Partitions)).andReturn(singletonMap(t1p0, offsetAndMetadata)); + consumer.seek(t1p0, offsetAndMetadata); + expectLastCall(); + replay(activeTaskCreator, topologyBuilder, consumer, changeLogReader); + taskManager.setPartitionResetter(tp -> assertThat(tp, is(empty()))); taskManager.handleAssignment(assignment, emptyMap()); + assertThat(nonRunningNonCorruptedTask.state(), is(Task.State.CREATED)); taskManager.handleCorruption(singletonMap(taskId00, taskId00Partitions)); - assertThat(nonRunningNonCorruptedTask.state(), is(Task.State.CREATED)); - assertThat(nonRunningNonCorruptedTask.partitionsForOffsetReset, equalTo(Collections.emptySet())); - assertThat(corruptedTask.partitionsForOffsetReset, equalTo(taskId00Partitions)); verify(activeTaskCreator); assertFalse(nonRunningNonCorruptedTask.commitPrepared); @@ -790,7 +814,7 @@ public Map prepareCommit() { replay(activeTaskCreator, standbyTaskCreator, topologyBuilder, consumer, changeLogReader); taskManager.handleAssignment(taskId01Assignment, taskId00Assignment); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); // make sure this will be committed and throw assertThat(runningNonCorruptedActive.state(), is(Task.State.RUNNING)); @@ -817,7 +841,7 @@ public void shouldCloseStandbyUnassignedTasksWhenCreatingNewTasks() { replay(activeTaskCreator, standbyTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(emptyMap(), taskId00Assignment); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); taskManager.handleAssignment(emptyMap(), emptyMap()); @@ -839,12 +863,12 @@ public void shouldAddNonResumedSuspendedTasks() { replay(activeTaskCreator, standbyTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(taskId00Assignment, taskId01Assignment); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); assertThat(task01.state(), is(Task.State.RUNNING)); taskManager.handleAssignment(taskId00Assignment, taskId01Assignment); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); assertThat(task01.state(), is(Task.State.RUNNING)); @@ -863,13 +887,13 @@ public void shouldUpdateInputPartitionsAfterRebalance() { taskManager.handleAssignment(taskId00Assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); final Set newPartitionsSet = mkSet(t1p1); final Map> taskIdSetMap = singletonMap(taskId00, newPartitionsSet); taskManager.handleAssignment(taskIdSetMap, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); assertEquals(newPartitionsSet, task00.inputPartitions()); verify(activeTaskCreator, consumer, changeLogReader); @@ -894,7 +918,7 @@ public void shouldAddNewActiveTasks() { assertThat(task00.state(), is(Task.State.CREATED)); - taskManager.tryToCompleteRestoration(time.milliseconds(), noOpResetter -> { }); + taskManager.tryToCompleteRestoration(time.milliseconds()); assertThat(task00.state(), is(Task.State.RUNNING)); assertThat(taskManager.activeTaskMap(), Matchers.equalTo(singletonMap(taskId00, task00))); @@ -938,7 +962,7 @@ public void initializeIfNeeded() { assertThat(task00.state(), is(Task.State.CREATED)); assertThat(task01.state(), is(Task.State.CREATED)); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(false)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(false)); assertThat(task00.state(), is(Task.State.CREATED)); assertThat(task01.state(), is(Task.State.CREATED)); @@ -957,7 +981,7 @@ public void shouldNotCompleteRestorationIfTaskCannotCompleteRestoration() { ); final Task task00 = new StateMachineTask(taskId00, taskId00Partitions, true) { @Override - public void completeRestoration(final java.util.function.Consumer> offsetResetter) { + public void completeRestoration() { throw new TimeoutException("timeout!"); } }; @@ -978,7 +1002,7 @@ public void completeRestoration(final java.util.function.Consumer taskManager.handleRevocation(taskId00Partitions)); @@ -1319,7 +1343,7 @@ public void closeDirty() { assertThat(task02.state(), is(Task.State.CREATED)); assertThat(task03.state(), is(Task.State.CREATED)); - taskManager.tryToCompleteRestoration(time.milliseconds(), null); + taskManager.tryToCompleteRestoration(time.milliseconds()); assertThat(task00.state(), is(Task.State.RESTORING)); assertThat(task01.state(), is(Task.State.RUNNING)); @@ -1388,7 +1412,7 @@ public Collection changelogPartitions() { assertThat(task00.state(), is(Task.State.CREATED)); - taskManager.tryToCompleteRestoration(time.milliseconds(), null); + taskManager.tryToCompleteRestoration(time.milliseconds()); assertThat(task00.state(), is(Task.State.RESTORING)); assertThat( @@ -1439,7 +1463,7 @@ public Collection changelogPartitions() { assertThat(task00.state(), is(Task.State.CREATED)); - taskManager.tryToCompleteRestoration(time.milliseconds(), null); + taskManager.tryToCompleteRestoration(time.milliseconds()); assertThat(task00.state(), is(Task.State.RESTORING)); assertThat( @@ -1574,7 +1598,7 @@ public void suspend() { assertThat(task01.state(), is(Task.State.CREATED)); assertThat(task02.state(), is(Task.State.CREATED)); - taskManager.tryToCompleteRestoration(time.milliseconds(), null); + taskManager.tryToCompleteRestoration(time.milliseconds()); assertThat(task00.state(), is(Task.State.RESTORING)); assertThat(task01.state(), is(Task.State.RUNNING)); @@ -1627,7 +1651,7 @@ public void shouldCloseStandbyTasksOnShutdown() { taskManager.handleAssignment(emptyMap(), assignment); assertThat(task00.state(), is(Task.State.CREATED)); - taskManager.tryToCompleteRestoration(time.milliseconds(), null); + taskManager.tryToCompleteRestoration(time.milliseconds()); assertThat(task00.state(), is(Task.State.RUNNING)); assertThat(taskManager.activeTaskMap(), Matchers.anEmptyMap()); assertThat(taskManager.standbyTaskMap(), Matchers.equalTo(singletonMap(taskId00, task00))); @@ -1649,7 +1673,7 @@ public void shouldInitializeNewActiveTasks() { replay(activeTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(taskId00Assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); assertThat(taskManager.activeTaskMap(), Matchers.equalTo(singletonMap(taskId00, task00))); @@ -1669,7 +1693,7 @@ public void shouldInitializeNewStandbyTasks() { replay(standbyTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(emptyMap(), taskId01Assignment); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task01.state(), is(Task.State.RUNNING)); assertThat(taskManager.activeTaskMap(), Matchers.anEmptyMap()); @@ -1709,7 +1733,7 @@ public void shouldCommitActiveAndStandbyTasks() { replay(activeTaskCreator, standbyTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(taskId00Assignment, taskId01Assignment); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); assertThat(task01.state(), is(Task.State.RUNNING)); @@ -1753,7 +1777,7 @@ public void shouldCommitProvidedTasksIfNeeded() { replay(activeTaskCreator, standbyTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(assignmentActive, assignmentStandby); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); assertThat(task01.state(), is(Task.State.RUNNING)); @@ -1784,7 +1808,7 @@ public void shouldNotCommitOffsetsIfOnlyStandbyTasksAssigned() { replay(activeTaskCreator, standbyTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(Collections.emptyMap(), taskId00Assignment); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); @@ -1810,7 +1834,7 @@ public void shouldNotCommitActiveAndStandbyTasksWhileRebalanceInProgress() throw replay(activeTaskCreator, standbyTaskCreator, stateDirectory, consumer, changeLogReader); taskManager.handleAssignment(taskId00Assignment, taskId01Assignment); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); assertThat(task01.state(), is(Task.State.RUNNING)); @@ -1923,7 +1947,7 @@ public Map prepareCommit() { replay(activeTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(taskId00Assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); @@ -1950,7 +1974,7 @@ public Map prepareCommit() { replay(standbyTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(emptyMap(), taskId01Assignment); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task01.state(), is(Task.State.RUNNING)); @@ -1985,7 +2009,7 @@ public Map purgeableOffsets() { replay(activeTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(taskId00Assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); @@ -2021,7 +2045,7 @@ public Map purgeableOffsets() { replay(activeTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(taskId00Assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); @@ -2053,7 +2077,7 @@ public void shouldIgnorePurgeDataErrors() { replay(activeTaskCreator, adminClient, consumer, changeLogReader); taskManager.handleAssignment(taskId00Assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); @@ -2105,7 +2129,7 @@ public void shouldMaybeCommitAllActiveTasksThatNeedCommit() { replay(activeTaskCreator, standbyTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(assignmentActive, assignmentStandby); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); assertThat(task01.state(), is(Task.State.RUNNING)); @@ -2145,7 +2169,7 @@ public void shouldProcessActiveTasks() { replay(activeTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); assertThat(task01.state(), is(Task.State.RUNNING)); @@ -2259,7 +2283,7 @@ public boolean process(final long wallClockTime) { replay(activeTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(taskId00Assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); @@ -2285,7 +2309,7 @@ public boolean process(final long wallClockTime) { replay(activeTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(taskId00Assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); @@ -2312,7 +2336,7 @@ public boolean maybePunctuateStreamTime() { replay(activeTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(taskId00Assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); @@ -2335,7 +2359,7 @@ public boolean maybePunctuateStreamTime() { replay(activeTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(taskId00Assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); @@ -2363,7 +2387,7 @@ public boolean maybePunctuateSystemTime() { replay(activeTaskCreator, consumer, changeLogReader); taskManager.handleAssignment(taskId00Assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); @@ -2387,7 +2411,7 @@ public Collection changelogPartitions() { replay(activeTaskCreator, changeLogReader, consumer); taskManager.handleAssignment(taskId00Assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(false)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(false)); assertThat(task00.state(), is(Task.State.RESTORING)); // this could be a bit mysterious; we're verifying _no_ interactions on the consumer, // since the taskManager should _not_ resume the assignment while we're still in RESTORING @@ -2409,7 +2433,7 @@ public void shouldHaveRemainingPartitionsUncleared() { try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(TaskManager.class)) { taskManager.handleAssignment(taskId00Assignment, emptyMap()); - assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); + assertThat(taskManager.tryToCompleteRestoration(time.milliseconds()), is(true)); assertThat(task00.state(), is(Task.State.RUNNING)); taskManager.handleRevocation(mkSet(t1p0, new TopicPartition("unknown", 0))); @@ -2565,7 +2589,7 @@ private Map handleAssignment(final Map allTasks = new HashMap<>(); @@ -2926,7 +2950,6 @@ private static class StateMachineTask extends AbstractTask implements Task { private Map committableOffsets = Collections.emptyMap(); private Map purgeableOffsets; private Map changelogOffsets = Collections.emptyMap(); - private Set partitionsForOffsetReset = Collections.emptySet(); private Long timeout = null; private final Map>> queue = new HashMap<>(); @@ -2956,12 +2979,7 @@ public void initializeIfNeeded() { } @Override - public void addPartitionsForOffsetReset(final Set partitionsForOffsetReset) { - this.partitionsForOffsetReset = partitionsForOffsetReset; - } - - @Override - public void completeRestoration(final java.util.function.Consumer> offsetResetter) { + public void completeRestoration() { if (state() == State.RUNNING) { return; } diff --git a/streams/src/test/resources/log4j.properties b/streams/src/test/resources/log4j.properties index eabbc549b93b6..f9ee64475888f 100644 --- a/streams/src/test/resources/log4j.properties +++ b/streams/src/test/resources/log4j.properties @@ -19,7 +19,7 @@ log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c:%L)%n log4j.logger.kafka=INFO -log4j.logger.org.apache.kafka=ERROR +log4j.logger.org.apache.kafka=INFO log4j.logger.org.apache.zookeeper=ERROR log4j.logger.org.apache.kafka.clients=WARN diff --git a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java index da3dd6f6ca7a1..5349d353ad253 100644 --- a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java +++ b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java @@ -529,7 +529,7 @@ private void setupTask(final StreamsConfig streamsConfig, context, logContext); task.initializeIfNeeded(); - task.completeRestoration(noOpResetter -> { }); + task.completeRestoration(); task.processorContext().setRecordContext(null); // initialize the task metadata so that all topics have zero lag