KAFKA-10029; Don't update completedReceives when channels are closed to avoid ConcurrentModificationException #8705

rajinisivaram · 2020-05-21T10:57:26Z

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

chia7712 · 2020-05-21T16:34:20Z

How about making a collection copy of completedReceives when traversing completedReceives?

rajinisivaram · 2020-05-21T17:32:08Z

@chia7712 I was tempted to do that initially, but that is not the pattern we use for everything else in Selector and it has always been this way (for several years), so adding tests to make sure we don't break it made more sense.

chia7712 · 2020-05-22T08:00:01Z

core/src/test/scala/unit/kafka/network/SocketServerTest.scala

@@ -1742,6 +1746,12 @@ class SocketServerTest {
           selector = Some(testableSelector)
           testableSelector
        }
+
+        override private[network] def processException(errorMessage: String, throwable: Throwable, isUncaught: Boolean): Unit = {


isUncaught is used by testing only so it is a bit awkward to production code. Could you check the errorMessage instead of adding new argument? for example:

if (errorMessage == "Processor got uncaught exception.") uncaughtExceptions += 1

Funny you should say that, I was initially checking the error message and then felt that the test wouldn't fail if the error message was changed. But since then I had also updated the test which triggers the uncaught exception code path, so it is actually safe now to check the error message. Have updated the code.

ijuma · 2020-05-25T21:49:06Z

@rajinisivaram What's the implication of not removing the completed receive in doClose?

…to avoid ConcurrentModificationException

rajinisivaram · 2020-05-26T10:09:23Z

@ijuma By not updating completedReceives when the channel is closed, we retain a reference to the channel until the next poll(). In terms of receives themselves, there is no impact. Looking at the code again, it seems unnecessary to use KafkaChannel as the key for the receive map. Like all other maps in Selector that store poll state for the channel, we can use the channel id instead. Updated the code.

ijuma · 2020-05-27T14:19:41Z

@rajinisivaram We also retain a reference to the NetworkReceive, which is probably a bigger deal, right?

ijuma · 2020-05-27T14:25:09Z

This change is probably OK, but the way we call close while iterating over completedReceives seems a bit fragile. It would probably be safer to collect the items we need to close and close them in a separate iteration. What do you think @rajinisivaram?

ijuma · 2020-05-27T14:26:09Z

One more thing, can we improve KafkaChannel.hashCode/equals to avoid unnecessary work? The calls to Objects.equals and Objects.hash seem pointless.

rajinisivaram · 2020-05-27T16:07:34Z

@ijuma Based on our discussion, I have added Selector#clearCompletedSends() and Selector#clearCompletedReceives() for SocketServer to clear buffers after they are processed. Also updated KafkaChannel. Can you review please? Thank you.

ijuma · 2020-05-28T04:57:56Z

core/src/test/scala/unit/kafka/network/SocketServerTest.scala

+          channelOpt.foreach { channel => map.put(channel.id, receive) }
+        }
+        cachedCompletedSends.currentPollValues.foreach(super.completedSends.add)
+        cachedDisconnected.currentPollValues.foreach { case (id, state) => super.disconnected.put(id, state) }


Can we add a comment explaining what we're trying to do here? It's not clear why we do (for example):

cachedCompletedSends.update(super.completedSends.asScala) followed by
cachedCompletedSends.currentPollValues.foreach(super.completedSends.add)

I have moved the second line that updates current result into update(). They were done separately because each type uses slightly different format, but it is clearer if they are together. Added comments as well.

ijuma · 2020-05-28T04:59:04Z

core/src/test/scala/unit/kafka/network/SocketServerTest.scala

@@ -1807,6 +1817,7 @@ class SocketServerTest {
          currentPollValues ++= newValues
        } else
          deferredValues ++= newValues
+        newValues.clear()


What is the goal of this?

We return minPerPoll results together, so the current values are cleared and then populated as necessary. The code is now in the same place, so hopefully that is clearer.

The reason I was asking is that we call toBuffer before calling this method for a couple of cases. That will basically create a copy, so then this clear won't do anything useful?

ah, I refactored a bit to clear the original buffer in each case.

ijuma

Thanks for the updates, looks good overall. I had some questions about the tests and one nit suggestion.

ijuma · 2020-05-28T05:04:53Z

clients/src/main/java/org/apache/kafka/common/network/Selector.java

-     * Clear the results from the prior poll
+     * Clears completed receives. This is used by SocketServer to remove references to
+     * receive buffers after processing completed receives, without waiting for the next
+     * poll() after all results have been processed.


Nit: maybe after all results have been processed is a bit redundant? Same for the clearCompletedSends docs.

rajinisivaram · 2020-05-28T09:21:03Z

@ijuma Thanks for the review, have addressed the comments.

ijuma · 2020-05-28T13:11:00Z

core/src/test/scala/unit/kafka/network/SocketServerTest.scala

-        cachedCompletedSends.update(super.completedSends.asScala)
-        cachedDisconnected.update(super.disconnected.asScala.toBuffer)
+
+        val map: util.Map[String, NetworkReceive] = JTestUtils.fieldValue(this, classOf[Selector], "completedReceives")


Should we call this completedReceivesMap or something?

@ijuma Thanks for the review, updated.

ijuma

LGTM, thanks. Just one minor suggestion below, no need for re-review.

ijuma · 2020-05-29T01:29:03Z

core/src/test/scala/unit/kafka/network/SocketServerTest.scala

-          val channelOpt = Option(super.channel(receive.source)).orElse(Option(super.closingChannel(receive.source)))
-          channelOpt.foreach { channel => completedReceivesMap.put(channel.id, receive) }
-        }
-
        // For each result type (completedReceives/completedSends/disconnected), defer the result to a subsequent poll()
        // if `minPerPoll` results are not yet available. When sufficient results are available, all available results
        // including previously deferred results are returned. This allows tests to process `minPerPoll` elements as the
        // results of a single poll iteration.


I think your refactoring has added the comments to other places. Maybe we can add "This allows tests to process minPerPoll elements as the results of a single poll iteration" to the update method and remove this?

rajinisivaram · 2020-05-29T08:32:06Z

@ijuma @chia7712 Thanks for the reviews, merging to trunk and 2.5.

…to avoid ConcurrentModificationException (#8705) Reviewers: Ismael Juma <ismael@juma.me.uk>, Chia-Ping Tsai <chia7712@gmail.com>

* 'trunk' of github.com:apache/kafka: (36 commits) Remove redundant `containsKey` call in KafkaProducer (apache#8761) KAFKA-9494; Include additional metadata information in DescribeConfig response (KIP-569) (apache#8723) KAFKA-10061; Fix flaky `ReassignPartitionsIntegrationTest.testCancellation` (apache#8749) KAFKA-9130; KIP-518 Allow listing consumer groups per state (apache#8238) KAFKA-9501: convert between active and standby without closing stores (apache#8248) KAFKA-10056; Ensure consumer metadata contains new topics on subscription change (apache#8739) MINOR: Log the reason for coordinator discovery failure (apache#8747) KAFKA-10029; Don't update completedReceives when channels are closed to avoid ConcurrentModificationException (apache#8705) MINOR: remove unnecessary timeout for admin request (apache#8738) MINOR: Relax Percentiles test (apache#8748) MINOR: regression test for task assignor config (apache#8743) MINOR: Update documentation.html to refer to 2.6 (apache#8745) MINOR: Update documentation.html to refer to 2.5 (apache#8744) KAFKA-9673: Filter and Conditional SMTs (apache#8699) KAFKA-9971: Error Reporting in Sink Connectors (KIP-610) (apache#8720) KAFKA-10052: Harden assertion of topic settings in Connect integration tests (apache#8735) MINOR: Slight MetadataCache tweaks to avoid unnecessary work (apache#8728) KAFKA-9802; Increase transaction timeout in system tests to reduce flakiness (apache#8736) KAFKA-10050: kafka_log4j_appender.py fixed for JDK11 (apache#8731) KAFKA-9146: Add option to force delete active members in StreamsResetter (apache#8589) ... # Conflicts: # core/src/main/scala/kafka/log/Log.scala

* apache-github/2.6: (32 commits) KAFKA-10083: fix failed testReassignmentWithRandomSubscriptionsAndChanges tests (apache#8786) KAFKA-9945: TopicCommand should support --if-exists and --if-not-exists when --bootstrap-server is used (apache#8737) KAFKA-9320: Enable TLSv1.3 by default (KIP-573) (apache#8695) KAFKA-10082: Fix the failed testMultiConsumerStickyAssignment (apache#8777) MINOR: Remove unused variable to fix spotBugs failure (apache#8779) MINOR: ChangelogReader should poll for duration 0 for standby restore (apache#8773) KAFKA-10030: Allow fetching a key from a single partition (apache#8706) Kafka-10064 Add documentation for KIP-571 (apache#8760) MINOR: Code cleanup and assertion message fixes in Connect integration tests (apache#8750) KAFKA-9987: optimize sticky assignment algorithm for same-subscription case (apache#8668) KAFKA-9392; Clarify deleteAcls javadoc and add test for create/delete timing (apache#7956) KAFKA-10074: Improve performance of `matchingAcls` (apache#8769) KAFKA-9494; Include additional metadata information in DescribeConfig response (KIP-569) (apache#8723) KAFKA-10056; Ensure consumer metadata contains new topics on subscription change (apache#8739) KAFKA-10029; Don't update completedReceives when channels are closed to avoid ConcurrentModificationException (apache#8705) KAFKA-10061; Fix flaky `ReassignPartitionsIntegrationTest.testCancellation` (apache#8749) KAFKA-9130; KIP-518 Allow listing consumer groups per state (apache#8238) KAFKA-9501: convert between active and standby without closing stores (apache#8248) MINOR: Relax Percentiles test (apache#8748) MINOR: regression test for task assignor config (apache#8743) ...

rajinisivaram requested a review from omkreddy May 21, 2020 10:58

chia7712 reviewed May 22, 2020

View reviewed changes

rajinisivaram added 2 commits May 26, 2020 10:57

KAFKA-10029; Don't update completedReceives when channels are closed …

74107fa

…to avoid ConcurrentModificationException

Address review comments

09589be

rajinisivaram force-pushed the KAFKA-10029-completed-receive branch from a551d91 to 09589be Compare May 26, 2020 10:02

rajinisivaram added 2 commits May 27, 2020 16:56

Address review comments

64f84b3

Address review comment

8cbe30e

rajinisivaram requested a review from ijuma May 27, 2020 16:07

ijuma reviewed May 28, 2020

View reviewed changes

Address review comments

a44db94

ijuma reviewed May 28, 2020

View reviewed changes

rajinisivaram added 2 commits May 28, 2020 14:16

Address review comment

ceace47

Address review comments

d572d8e

ijuma approved these changes May 29, 2020

View reviewed changes

Address review comment

e605e29

rajinisivaram merged commit 1fd195e into apache:trunk May 29, 2020

rajinisivaram added a commit that referenced this pull request May 29, 2020

KAFKA-10029; Don't update completedReceives when channels are closed …

9ee465f

…to avoid ConcurrentModificationException (#8705) Reviewers: Ismael Juma <ismael@juma.me.uk>, Chia-Ping Tsai <chia7712@gmail.com>

rajinisivaram added a commit that referenced this pull request May 29, 2020

KAFKA-10029; Don't update completedReceives when channels are closed …

6fee20b

…to avoid ConcurrentModificationException (#8705) Reviewers: Ismael Juma <ismael@juma.me.uk>, Chia-Ping Tsai <chia7712@gmail.com>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-10029; Don't update completedReceives when channels are closed to avoid ConcurrentModificationException #8705

KAFKA-10029; Don't update completedReceives when channels are closed to avoid ConcurrentModificationException #8705

rajinisivaram commented May 21, 2020

chia7712 commented May 21, 2020

rajinisivaram commented May 21, 2020

chia7712 May 22, 2020

rajinisivaram May 26, 2020

ijuma commented May 25, 2020

rajinisivaram commented May 26, 2020

ijuma commented May 27, 2020

ijuma commented May 27, 2020

ijuma commented May 27, 2020

rajinisivaram commented May 27, 2020

ijuma May 28, 2020

rajinisivaram May 28, 2020

ijuma May 28, 2020

rajinisivaram May 28, 2020

ijuma May 28, 2020

rajinisivaram May 28, 2020

ijuma left a comment

ijuma May 28, 2020

rajinisivaram commented May 28, 2020

ijuma May 28, 2020

rajinisivaram May 28, 2020

ijuma left a comment

ijuma May 29, 2020

rajinisivaram commented May 29, 2020

KAFKA-10029; Don't update completedReceives when channels are closed to avoid ConcurrentModificationException #8705

KAFKA-10029; Don't update completedReceives when channels are closed to avoid ConcurrentModificationException #8705

Conversation

rajinisivaram commented May 21, 2020

Committer Checklist (excluded from commit message)

chia7712 commented May 21, 2020

rajinisivaram commented May 21, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ijuma commented May 25, 2020

rajinisivaram commented May 26, 2020

ijuma commented May 27, 2020

ijuma commented May 27, 2020

ijuma commented May 27, 2020

rajinisivaram commented May 27, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ijuma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rajinisivaram commented May 28, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ijuma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rajinisivaram commented May 29, 2020