Repository: kafka Updated Branches: refs/heads/trunk b8c3b7c96 -> b22a4f078
KAFKA-2000; Delete topic should also delete consumer offsets. Author: Manikumar Reddy O <[email protected]> Reviewers: Jason Gustafson <[email protected]>, Sriharsha Chintalapani <[email protected]> Closes #1850 from omkreddy/KAFKA-2700-DELETE Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/b22a4f07 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/b22a4f07 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/b22a4f07 Branch: refs/heads/trunk Commit: b22a4f0782d0f4d3e07075403e8e5dc318d8d387 Parents: b8c3b7c Author: Manikumar Reddy O <[email protected]> Authored: Wed Jan 25 08:17:50 2017 -0800 Committer: Sriharsha Chintalapani <[email protected]> Committed: Wed Jan 25 08:17:50 2017 -0800 ---------------------------------------------------------------------- .../kafka/coordinator/GroupCoordinator.scala | 4 +++ .../scala/kafka/coordinator/GroupMetadata.scala | 21 ++++++++++++-- .../coordinator/GroupMetadataManager.scala | 12 ++++++-- .../src/main/scala/kafka/server/KafkaApis.scala | 3 +- .../main/scala/kafka/server/MetadataCache.scala | 8 ++++-- .../scala/kafka/server/ReplicaManager.scala | 5 ++-- .../unit/kafka/server/OffsetCommitTest.scala | 29 +++++++++++++++++++- 7 files changed, 71 insertions(+), 11 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/b22a4f07/core/src/main/scala/kafka/coordinator/GroupCoordinator.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/coordinator/GroupCoordinator.scala b/core/src/main/scala/kafka/coordinator/GroupCoordinator.scala index 7abbc6e..c7edcef 100644 --- a/core/src/main/scala/kafka/coordinator/GroupCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/GroupCoordinator.scala @@ -497,6 +497,10 @@ class GroupCoordinator(val brokerId: Int, } } + def handleDeletedPartitions(topicPartitions: Seq[TopicPartition]) { + groupManager.cleanupGroupMetadata(Some(topicPartitions)) + } + private def onGroupUnloaded(group: GroupMetadata) { group synchronized { info(s"Unloading group metadata for ${group.groupId} with generation ${group.generationId}") http://git-wip-us.apache.org/repos/asf/kafka/blob/b22a4f07/core/src/main/scala/kafka/coordinator/GroupMetadata.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/coordinator/GroupMetadata.scala b/core/src/main/scala/kafka/coordinator/GroupMetadata.scala index 6b889f4..eac3151 100644 --- a/core/src/main/scala/kafka/coordinator/GroupMetadata.scala +++ b/core/src/main/scala/kafka/coordinator/GroupMetadata.scala @@ -17,8 +17,10 @@ package kafka.coordinator -import collection.mutable +import collection.{Seq, mutable, immutable} + import java.util.UUID + import kafka.common.OffsetAndMetadata import kafka.utils.nonthreadsafe import org.apache.kafka.common.TopicPartition @@ -264,7 +266,9 @@ private[coordinator] class GroupMetadata(val groupId: String, initialState: Grou } def completePendingOffsetWrite(topicPartition: TopicPartition, offset: OffsetAndMetadata) { - offsets.put(topicPartition, offset) + if (pendingOffsetCommits.contains(topicPartition)) + offsets.put(topicPartition, offset) + pendingOffsetCommits.get(topicPartition) match { case Some(stagedOffset) if offset == stagedOffset => pendingOffsetCommits.remove(topicPartition) case _ => @@ -282,12 +286,23 @@ private[coordinator] class GroupMetadata(val groupId: String, initialState: Grou pendingOffsetCommits ++= offsets } + def removeOffsets(topicPartitions: Seq[TopicPartition]): immutable.Map[TopicPartition, OffsetAndMetadata] = { + val removedOffsetMap = new mutable.HashMap[TopicPartition, OffsetAndMetadata] + for (topicPartition <- topicPartitions) { + pendingOffsetCommits.remove(topicPartition) + val removedOffsets = offsets.remove(topicPartition) + if (!removedOffsets.isEmpty) + removedOffsetMap.put(topicPartition, removedOffsets.get) + } + removedOffsetMap.toMap + } + def removeExpiredOffsets(startMs: Long) = { val expiredOffsets = offsets.filter { case (topicPartition, offset) => offset.expireTimestamp < startMs && !pendingOffsetCommits.contains(topicPartition) } offsets --= expiredOffsets.keySet - expiredOffsets + expiredOffsets.toMap } def allOffsets = offsets.toMap http://git-wip-us.apache.org/repos/asf/kafka/blob/b22a4f07/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala b/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala index 2d6889c..1163e72 100644 --- a/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala +++ b/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala @@ -546,14 +546,22 @@ class GroupMetadataManager(val brokerId: Int, } // visible for testing - private[coordinator] def cleanupGroupMetadata() { + private[coordinator] def cleanupGroupMetadata(): Unit = { + cleanupGroupMetadata(None) + } + + def cleanupGroupMetadata(topicPartitions: Option[Seq[TopicPartition]]) { val startMs = time.milliseconds() var offsetsRemoved = 0 groupMetadataCache.foreach { case (groupId, group) => val (expiredOffsets, groupIsDead, generation) = group synchronized { // remove expired offsets from the cache - val expiredOffsets = group.removeExpiredOffsets(startMs) + val expiredOffsets = if (topicPartitions.isEmpty) + group.removeExpiredOffsets(startMs) + else + group.removeOffsets(topicPartitions.get) + if (group.is(Empty) && !group.hasOffsets) { info(s"Group $groupId transitioned to Dead in generation ${group.generationId}") group.transitionTo(Dead) http://git-wip-us.apache.org/repos/asf/kafka/blob/b22a4f07/core/src/main/scala/kafka/server/KafkaApis.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index 530dafc..b99ce4d 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/core/src/main/scala/kafka/server/KafkaApis.scala @@ -196,7 +196,8 @@ class KafkaApis(val requestChannel: RequestChannel, val updateMetadataResponse = if (authorize(request.session, ClusterAction, Resource.ClusterResource)) { - replicaManager.maybeUpdateMetadataCache(correlationId, updateMetadataRequest, metadataCache) + val deletedPartitions = replicaManager.maybeUpdateMetadataCache(correlationId, updateMetadataRequest, metadataCache) + coordinator.handleDeletedPartitions(deletedPartitions) if (adminManager.hasDelayedTopicOperations) { updateMetadataRequest.partitionStates.keySet.asScala.map(_.topic).foreach { topic => adminManager.tryCompleteDelayedTopicOperations(topic) http://git-wip-us.apache.org/repos/asf/kafka/blob/b22a4f07/core/src/main/scala/kafka/server/MetadataCache.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/server/MetadataCache.scala b/core/src/main/scala/kafka/server/MetadataCache.scala index 5c28e14..9a6090d 100755 --- a/core/src/main/scala/kafka/server/MetadataCache.scala +++ b/core/src/main/scala/kafka/server/MetadataCache.scala @@ -27,7 +27,7 @@ import kafka.common.{BrokerEndPointNotAvailableException, Topic, TopicAndPartiti import kafka.controller.{KafkaController, LeaderIsrAndControllerEpoch} import kafka.utils.CoreUtils._ import kafka.utils.Logging -import org.apache.kafka.common.Node +import org.apache.kafka.common.{Node, TopicPartition} import org.apache.kafka.common.network.ListenerName import org.apache.kafka.common.protocol.Errors import org.apache.kafka.common.requests.{MetadataResponse, PartitionState, UpdateMetadataRequest} @@ -155,7 +155,8 @@ private[server] class MetadataCache(brokerId: Int) extends Logging { def getControllerId: Option[Int] = controllerId - def updateCache(correlationId: Int, updateMetadataRequest: UpdateMetadataRequest) { + // This method returns the deleted TopicPartitions received from UpdateMetadataRequest + def updateCache(correlationId: Int, updateMetadataRequest: UpdateMetadataRequest): Seq[TopicPartition] = { inWriteLock(partitionMetadataLock) { controllerId = updateMetadataRequest.controllerId match { case id if id < 0 => None @@ -177,6 +178,7 @@ private[server] class MetadataCache(brokerId: Int) extends Logging { aliveNodes(broker.id) = nodes.asScala } + val deletedPartitions = new mutable.ArrayBuffer[TopicPartition] updateMetadataRequest.partitionStates.asScala.foreach { case (tp, info) => val controllerId = updateMetadataRequest.controllerId val controllerEpoch = updateMetadataRequest.controllerEpoch @@ -184,6 +186,7 @@ private[server] class MetadataCache(brokerId: Int) extends Logging { removePartitionInfo(tp.topic, tp.partition) stateChangeLogger.trace(s"Broker $brokerId deleted partition $tp from metadata cache in response to UpdateMetadata " + s"request sent by controller $controllerId epoch $controllerEpoch with correlation id $correlationId") + deletedPartitions += tp } else { val partitionInfo = partitionStateToPartitionStateInfo(info) addOrUpdatePartitionInfo(tp.topic, tp.partition, partitionInfo) @@ -191,6 +194,7 @@ private[server] class MetadataCache(brokerId: Int) extends Logging { s"UpdateMetadata request sent by controller $controllerId epoch $controllerEpoch with correlation id $correlationId") } } + deletedPartitions } } http://git-wip-us.apache.org/repos/asf/kafka/blob/b22a4f07/core/src/main/scala/kafka/server/ReplicaManager.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index 199bb67..be2e04a 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -625,7 +625,7 @@ class ReplicaManager(val config: KafkaConfig, replica.log.map(log => (log.config.messageFormatVersion.messageFormatVersion, log.config.messageTimestampType)) } - def maybeUpdateMetadataCache(correlationId: Int, updateMetadataRequest: UpdateMetadataRequest, metadataCache: MetadataCache) { + def maybeUpdateMetadataCache(correlationId: Int, updateMetadataRequest: UpdateMetadataRequest, metadataCache: MetadataCache) : Seq[TopicPartition] = { replicaStateChangeLock synchronized { if(updateMetadataRequest.controllerEpoch < controllerEpoch) { val stateControllerEpochErrorMessage = ("Broker %d received update metadata request with correlation id %d from an " + @@ -635,8 +635,9 @@ class ReplicaManager(val config: KafkaConfig, stateChangeLogger.warn(stateControllerEpochErrorMessage) throw new ControllerMovedException(stateControllerEpochErrorMessage) } else { - metadataCache.updateCache(correlationId, updateMetadataRequest) + val deletedPartitions = metadataCache.updateCache(correlationId, updateMetadataRequest) controllerEpoch = updateMetadataRequest.controllerEpoch + deletedPartitions } } } http://git-wip-us.apache.org/repos/asf/kafka/blob/b22a4f07/core/src/test/scala/unit/kafka/server/OffsetCommitTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/server/OffsetCommitTest.scala b/core/src/test/scala/unit/kafka/server/OffsetCommitTest.scala index 6ca784a..355acde 100755 --- a/core/src/test/scala/unit/kafka/server/OffsetCommitTest.scala +++ b/core/src/test/scala/unit/kafka/server/OffsetCommitTest.scala @@ -30,6 +30,8 @@ import org.junit.Assert._ import java.util.Properties import java.io.File +import kafka.admin.AdminUtils + import scala.util.Random import scala.collection._ @@ -46,7 +48,7 @@ class OffsetCommitTest extends ZooKeeperTestHarness { @Before override def setUp() { super.setUp() - val config: Properties = createBrokerConfig(1, zkConnect) + val config: Properties = createBrokerConfig(1, zkConnect, enableDeleteTopic = true) config.setProperty(KafkaConfig.OffsetsTopicReplicationFactorProp, "1") config.setProperty(KafkaConfig.OffsetsRetentionCheckIntervalMsProp, retentionCheckInterval.toString) val logDirPath = config.getProperty("log.dir") @@ -304,4 +306,29 @@ class OffsetCommitTest extends ZooKeeperTestHarness { assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION.code, commitResponse.commitStatus.get(TopicAndPartition(topic1, 0)).get) assertEquals(Errors.NONE.code, commitResponse.commitStatus.get(TopicAndPartition(topic2, 0)).get) } + + @Test + def testOffsetsDeleteAfterTopicDeletion() { + // set up topic partition + val topic = "topic" + val topicPartition = TopicAndPartition(topic, 0) + createTopic(zkUtils, topic, servers = Seq(server), numPartitions = 1) + + val commitRequest = OffsetCommitRequest(group, immutable.Map(topicPartition -> OffsetAndMetadata(offset = 42L))) + val commitResponse = simpleConsumer.commitOffsets(commitRequest) + + assertEquals(Errors.NONE.code, commitResponse.commitStatus.get(topicPartition).get) + + // start topic deletion + AdminUtils.deleteTopic(zkUtils, topic) + TestUtils.verifyTopicDeletion(zkUtils, topic, 1, Seq(server)) + Thread.sleep(retentionCheckInterval * 2) + + // check if offsets deleted + val fetchRequest = OffsetFetchRequest(group, Seq(TopicAndPartition(topic, 0))) + val offsetMetadataAndErrorMap = simpleConsumer.fetchOffsets(fetchRequest) + val offsetMetadataAndError = offsetMetadataAndErrorMap.requestInfo(topicPartition) + assertEquals(OffsetMetadataAndError.NoOffset, offsetMetadataAndError) + } + }
