Skip to content

KAFKA-18384 Remove ZkAlterPartitionManager #18364

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mimaison merged 4 commits into from
Jan 7, 2025
Merged

KAFKA-18384 Remove ZkAlterPartitionManager #18364

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
59a4911
remove ZkAlterPartitionManager
m1a2st Dec 31, 2024
9f2c5ed
Merge branch 'trunk' into KAFKA-18384
m1a2st Jan 6, 2025
70d4d31
addressed by comments
m1a2st Jan 6, 2025
80fd87f
addressed by comments
m1a2st Jan 7, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 0 additions & 12 deletions core/src/main/scala/kafka/server/AlterPartitionManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@ import java.util
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.{CompletableFuture, ConcurrentHashMap}
import kafka.utils.Logging
import kafka.zk.KafkaZkClient
import org.apache.kafka.clients.ClientResponse
import org.apache.kafka.common.TopicIdPartition
import org.apache.kafka.common.TopicPartition
Expand Down Expand Up @@ -100,17 +99,6 @@ object AlterPartitionManager {
metadataVersionSupplier = () => metadataCache.metadataVersion()
)
}

/**
* Factory for ZK based implementation, used when IBP < 2.7-IV2
*/
def apply(
scheduler: Scheduler,
time: Time,
zkClient: KafkaZkClient
): AlterPartitionManager = {
new ZkAlterPartitionManager(scheduler, time, zkClient)
}
}

class DefaultAlterPartitionManager(
Expand Down
111 changes: 0 additions & 111 deletions core/src/main/scala/kafka/server/ZkAlterPartitionManager.scala
View file
Open in desktop

This file was deleted.

69 changes: 1 addition & 68 deletions core/src/test/scala/unit/kafka/cluster/PartitionTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@ import com.yammer.metrics.core.Metric
import kafka.log._
import kafka.server._
import kafka.utils._
import kafka.zk.KafkaZkClient
import org.apache.kafka.common.errors.{ApiException, FencedLeaderEpochException, InconsistentTopicIdException, InvalidTxnStateException, NotLeaderOrFollowerException, OffsetNotAvailableException, OffsetOutOfRangeException, UnknownLeaderEpochException}
import org.apache.kafka.common.message.{AlterPartitionResponseData, FetchResponseData}
import org.apache.kafka.common.message.LeaderAndIsrRequestData.LeaderAndIsrPartitionState
Expand All @@ -36,7 +35,7 @@ import org.apache.kafka.metadata.LeaderRecoveryState
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Test
import org.mockito.ArgumentMatchers
import org.mockito.ArgumentMatchers.{any, anyBoolean, anyInt, anyLong, anyString}
import org.mockito.ArgumentMatchers.{any, anyBoolean, anyInt, anyLong}

@mimaison mimaison Jan 6, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you also remove the ZooKeeper bits in testIsrNotExpandedIfReplicaIsFencedOrShutdown()?

import org.mockito.Mockito._
import org.mockito.invocation.InvocationOnMock

Expand All @@ -55,7 +54,6 @@ import org.apache.kafka.common.replica.ClientMetadata.DefaultClientMetadata
import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
import org.apache.kafka.coordinator.transaction.TransactionLogConfig
import org.apache.kafka.server.common.{ControllerRequestCompletionHandler, MetadataVersion, NodeToControllerChannelManager, RequestLocal}
import org.apache.kafka.server.common.MetadataVersion.IBP_2_6_IV0
import org.apache.kafka.server.metrics.KafkaYammerMetrics
import org.apache.kafka.server.purgatory.{DelayedOperationPurgatory, TopicPartitionOperationKey}
import org.apache.kafka.server.share.fetch.DelayedShareFetchPartitionKey
Expand Down Expand Up @@ -2698,71 +2696,6 @@ class PartitionTest extends AbstractPartitionTest {
assertEquals(alterPartitionManager.isrUpdates.size, 1)
}

@Test
def testZkIsrManagerAsyncCallback(): Unit = {
// We need a real scheduler here so that the ISR write lock works properly
val scheduler = new KafkaScheduler(1, true, "zk-isr-test")
scheduler.startup()
val kafkaZkClient = mock(classOf[KafkaZkClient])

doAnswer(_ => (true, 2))
.when(kafkaZkClient)
.conditionalUpdatePath(anyString(), any(), ArgumentMatchers.eq(1), any())

val zkIsrManager = AlterPartitionManager(scheduler, time, kafkaZkClient)
zkIsrManager.start()

val partition = new Partition(topicPartition,
replicaLagTimeMaxMs = ReplicationConfigs.REPLICA_LAG_TIME_MAX_MS_DEFAULT,
interBrokerProtocolVersion = IBP_2_6_IV0, // shouldn't matter, but set this to a ZK isr version
localBrokerId = brokerId,
() => defaultBrokerEpoch(brokerId),
time,
alterPartitionListener,
delayedOperations,
metadataCache,
logManager,
zkIsrManager)

val log = logManager.getOrCreateLog(topicPartition, topicId = None)
seedLogData(log, numRecords = 10, leaderEpoch = 4)

val controllerEpoch = 0
val leaderEpoch = 5
val follower1 = brokerId + 1
val follower2 = brokerId + 2
val follower3 = brokerId + 3
val replicas = Seq(brokerId, follower1, follower2, follower3)
val isr = Seq(brokerId, follower1, follower2)

doNothing().when(delayedOperations).checkAndCompleteAll()

assertTrue(makeLeader(
partition = partition,
topicId = None,
controllerEpoch = controllerEpoch,
leaderEpoch = leaderEpoch,
isr = isr,
replicas = replicas,
partitionEpoch = 1,
isNew = true
))
assertEquals(0L, partition.localLogOrException.highWatermark)

// Expand ISR
fetchFollower(partition, replicaId = follower3, fetchOffset = 10L)

// Try avoiding a race
TestUtils.waitUntilTrue(() => !partition.partitionState.isInflight, "Expected ISR state to be committed", 100)

partition.partitionState match {
case CommittedPartitionState(isr, _) => assertEquals(Set(brokerId, follower1, follower2, follower3), isr)
case _ => fail("Expected a committed ISR following Zk expansion")
}

scheduler.shutdown()
}

@Test
def testUseCheckpointToInitializeHighWatermark(): Unit = {
val log = logManager.getOrCreateLog(topicPartition, topicId = None)
Expand Down
32 changes: 2 additions & 30 deletions core/src/test/scala/unit/kafka/server/AlterPartitionManagerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,11 +19,10 @@ package kafka.server

import java.util.Collections
import java.util.stream.{Stream => JStream}
import kafka.zk.KafkaZkClient
import org.apache.kafka.clients.ClientResponse
import org.apache.kafka.common.TopicIdPartition
import org.apache.kafka.common.Uuid
import org.apache.kafka.common.errors.{AuthenticationException, InvalidUpdateVersionException, OperationNotAttemptedException, UnknownServerException, UnsupportedVersionException}
import org.apache.kafka.common.errors.{AuthenticationException, OperationNotAttemptedException, UnknownServerException, UnsupportedVersionException}
import org.apache.kafka.common.message.AlterPartitionRequestData.BrokerState
import org.apache.kafka.common.message.{AlterPartitionRequestData, AlterPartitionResponseData}
import org.apache.kafka.common.metrics.Metrics
Expand All @@ -43,7 +42,7 @@ import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.Arguments
import org.junit.jupiter.params.provider.MethodSource
import org.mockito.ArgumentMatcher
import org.mockito.ArgumentMatchers.{any, anyString}
import org.mockito.ArgumentMatchers.any
import org.mockito.Mockito.{mock, reset, times, verify}
import org.mockito.{ArgumentCaptor, ArgumentMatchers, Mockito}

Expand Down Expand Up @@ -629,33 +628,6 @@ class AlterPartitionManagerTest {
.setErrorCode(error.code)))
}

@Test
def testZkBasic(): Unit = {
val scheduler = new MockScheduler(time)
scheduler.startup()

val kafkaZkClient = Mockito.mock(classOf[KafkaZkClient])
Mockito.doAnswer(_ => (true, 2))
.when(kafkaZkClient)
.conditionalUpdatePath(anyString(), any(), ArgumentMatchers.eq(1), any())
Mockito.doAnswer(_ => (false, 2))
.when(kafkaZkClient)
.conditionalUpdatePath(anyString(), any(), ArgumentMatchers.eq(3), any())

val zkIsrManager = new ZkAlterPartitionManager(scheduler, time, kafkaZkClient)
zkIsrManager.start()

// Correct ZK version
val future1 = zkIsrManager.submit(tp0, new LeaderAndIsr(1, 1, List(1, 2, 3).map(Int.box).asJava, LeaderRecoveryState.RECOVERED, 1), 0)
assertTrue(future1.isDone)
assertEquals(new LeaderAndIsr(1, 1, List(1, 2, 3).map(Int.box).asJava, LeaderRecoveryState.RECOVERED, 2), future1.get)

// Wrong ZK version
val future2 = zkIsrManager.submit(tp0, new LeaderAndIsr(1, 1, List(1, 2, 3).map(Int.box).asJava, LeaderRecoveryState.RECOVERED, 3), 0)
assertTrue(future2.isCompletedExceptionally)
assertFutureThrows(future2, classOf[InvalidUpdateVersionException])
}

private def partitionResponse(
tp: TopicIdPartition = tp0,
error: Errors = Errors.NONE,
Expand Down