KAFKA-10301: Do not clear Partition#remoteReplicasMap during partition assignment updates by stanislavkozlovski · Pull Request #9065 · apache/kafka

stanislavkozlovski · 2020-07-23T15:14:05Z

We would previously update the map by adding the new replicas to the map and then removing the old ones. During a recent refactoring, we changed the logic to first clear the map and then add all the replicas to it.

While this is done in a write lock, not all callers that access the map structure use a lock. It is safer to revert to the previous behavior of showing the intermediate state of the map with extra replicas, rather than an intermediate state of the map with no replicas.

…n assignment updates

stanislavkozlovski · 2020-07-23T15:14:32Z

cc @ijuma

stanislavkozlovski · 2020-07-23T15:15:04Z

      .filter(_ != localBrokerId)
      .foreach(id => remoteReplicasMap.getAndMaybePut(id, new Replica(id, topicPartition)))
-
+    removedReplicas.foreach(remoteReplicasMap.remove)


I decided to not get fancy with refactorings - this is literally the old code (

kafka/core/src/main/scala/kafka/cluster/Partition.scala

Line 657 in 7f9187f

removedReplicas.foreach(remoteReplicasMap.remove)

))

Would remoteReplicasMap --= removedReplicas work here?

Oh, this is a Pool, so we would have to add a removeAll method. Seems easy enough though since it can call the relevant method in ConcurrentMap.

remoteReplicasMap --= removedReplicas doesn't compile - the remoteReplicasMap is using a Kafka Pool class which itself is using a Java Map and I don't think they support the --= notation

ijuma · 2020-07-23T15:22:42Z

ok to test

The given method is deprecated in Scala 2.13

stanislavkozlovski · 2020-07-23T15:56:35Z

Currently working on introducing a test case for this

stanislavkozlovski · 2020-07-23T17:05:26Z

+   * where replicas present both in the old and new assignment are missing
+   */
+  @Test
+  def testGetReplicaWithUpdateAssignmentAndIsr(): Unit = {


This fails incredibly quickly 100/100 times without the Partition.scala changes.

ijuma · 2020-07-23T17:20:04Z

ok to test

ijuma · 2020-07-23T17:20:53Z

+ * limitations under the License.
+ */
+
+package unit.kafka.utils


Remove unit.

ijuma

Thanks for the PR. A few minor comments below.

ijuma · 2020-07-23T17:31:08Z


  def remove(key: K, value: V): Boolean = pool.remove(key, value)

+  def removeAll(keys: Iterable[K]): Unit = pool.keySet().removeAll(keys.asJavaCollection)


Nit: () is not needed.

ijuma · 2020-07-23T17:34:40Z

+    val replicaToCheck = 3
+    val firstReplicaSet = Seq[Integer](3, 4, 5).asJava
+    val secondReplicaSet = Seq[Integer](1, 2, 3).asJava
+    def partitionState(replicas: java.util.List[Integer]): LeaderAndIsrPartitionState = new LeaderAndIsrPartitionState()


No need to repeat LeaderAndIsrPartitionState twice.

ijuma · 2020-07-23T17:36:03Z

+    partition.makeLeader(partitionState(secondReplicaSet), offsetCheckpoints)
+    assertTrue(s"Expected replica $replicaToCheck to be defined", partition.getReplica(replicaToCheck).isDefined)
+
+    var i = 0


Shouldn't this be inside the thread state?

Yeah, nice catch

ijuma · 2020-07-23T17:37:57Z

+      }
+    }): Runnable)
+
+    val deadline = 5.seconds.fromNow


5 seconds is quite a bit. Can it be lower?

I think so. I opted for 5s as I saw the other tests had up to 15s of waits for futures. Let me see if 1s can go

Lowered to 1s

ijuma · 2020-07-23T17:42:21Z

+    }): Runnable)
+
+    val deadline = 5.seconds.fromNow
+    while(deadline.hasTimeLeft()) {


Nit: space missing after while.

ijuma · 2020-07-24T03:49:29Z

ok to test

ijuma · 2020-07-24T13:15:40Z

retest this please

ijuma

LGTM, thanks. Will wait for the tests to pass.

ijuma · 2020-07-24T13:19:21Z

ok to test

ijuma · 2020-07-24T13:19:41Z

retest this please

ijuma · 2020-07-24T13:25:31Z

One PR build was started here:

https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3515/

ijuma · 2020-07-24T13:26:16Z

Looks like the last message somehow made the Jenkins status to be updated in the PR again.

ijuma · 2020-07-24T19:02:23Z

One build passed, one failed due to environmental reasons, one had a single flaky failure:

org.apache.kafka.connect.mirror.MirrorConnectorsIntegrationTest.testOneWayReplicationWithAutorOffsetSync1

I ran the full test suite locally and it passed as well. Merging to trunk and 2.6. cc @rhauch

…n assignment updates (#9065) We would previously update the map by adding the new replicas to the map and then removing the old ones. During a recent refactoring, we changed the logic to first clear the map and then add all the replicas to it. While this is done in a write lock, not all callers that access the map structure use a lock. It is safer to revert to the previous behavior of showing the intermediate state of the map with extra replicas, rather than an intermediate state of the map with no replicas. Reviewers: Ismael Juma <ismael@juma.me.uk>

…t-for-generated-requests * apache-github/trunk: (148 commits) MINOR: remove NewTopic#NO_PARTITIONS and NewTopic#NO_REPLICATION_FACTOR as they are duplicate to CreateTopicsRequest#NO_NUM_PARTITIONS and CreateTopicsRequest#NO_REPLICATION_FACTOR (apache#9077) MINOR: Remove staticmethod tag to be able to use logger of instance (apache#9086) MINOR: Adjust 'release.py' script to use shell when using gradlewAll and PGP signing, which were required to build the 2.6.0 RCs (apache#9045) MINOR: Update dependencies for Kafka 2.7 (part 1) (apache#9082) MINOR: INFO log4j when request re-join (apache#9068) MINOR: Recommend Java 11 (apache#9080) KAFKA-10306: GlobalThread should fail on InvalidOffsetException (apache#9075) KAFKA-10158: Fix flaky testDescribeUnderReplicatedPartitionsWhenReassignmentIsInProgress (apache#9022) MINOR: code cleanup for `VOut` inconsistent naming (apache#8907) KAFKA-10246 : AbstractProcessorContext topic() throws NPE (apache#9034) KAFKA-10305: Print usage when parsing fails for ConsumerPerformance (apache#9071) MINOR: removed incorrect deprecation annotations (apache#9061) MINOR: speed up release script (apache#9070) MINOR: add task ':streams:testAll' (apache#9073) KAFKA-10301: Do not clear Partition#remoteReplicasMap during partition assignment updates (apache#9065) KAFKA-10268: dynamic config like "--delete-config log.retention.ms" doesn't work (apache#9051) KAFKA-10300 fix flaky core/group_mode_transactions_test.py (apache#9059) MINOR: Publish metrics package in the javadoc (apache#9036) KAFKA-8264: decrease the record size for flaky test KAFKA-5876: Add new exception types for Interactive Queries (apache#8200) ...

KAFKA-10301: Do not clear Partition#remoteReplicasMap during partitio…

5611fd8

…n assignment updates

stanislavkozlovski commented Jul 23, 2020

View reviewed changes

Avoid using SetOps -- trait method

664b7cd

The given method is deprecated in Scala 2.13

stanislavkozlovski added 2 commits July 23, 2020 20:02

Prefer Pool#getAndMaybePut to avoid needless Replica instantiations

f6d9340

Add test to catch the race condition

12016f1

stanislavkozlovski commented Jul 23, 2020

View reviewed changes

Introduce Pool#removeAll helper method to improve readability

21fb575

ijuma reviewed Jul 23, 2020

View reviewed changes

Move PoolTest to same package as kafka.utils.Pool

c13e738

ijuma reviewed Jul 23, 2020

View reviewed changes

Reduce test timeout and address stylistic comments

0f813d6

ijuma approved these changes Jul 24, 2020

View reviewed changes

ijuma merged commit fa6e5b8 into apache:trunk Jul 24, 2020


		def remove(key: K, value: V): Boolean = pool.remove(key, value)

		def removeAll(keys: Iterable[K]): Unit = pool.keySet().removeAll(keys.asJavaCollection)

Conversation

stanislavkozlovski commented Jul 23, 2020

Uh oh!

stanislavkozlovski commented Jul 23, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ijuma commented Jul 23, 2020

Uh oh!

stanislavkozlovski commented Jul 23, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ijuma commented Jul 23, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ijuma left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ijuma commented Jul 24, 2020

Uh oh!

ijuma commented Jul 24, 2020

Uh oh!

ijuma left a comment

Choose a reason for hiding this comment

Uh oh!

ijuma commented Jul 24, 2020

Uh oh!

ijuma commented Jul 24, 2020

Uh oh!

ijuma commented Jul 24, 2020

Uh oh!

ijuma commented Jul 24, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ijuma commented Jul 24, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

ijuma commented Jul 24, 2020 •

edited

Loading