KAFKA-9796; Broker shutdown could be stuck forever under certain conditions by dajac · Pull Request #8448 · apache/kafka

dajac · 2020-04-08T12:24:27Z

This patch reworks the SocketServer to always start the acceptor threads after the processor threads and to always stop the acceptor threads before the processor threads. It ensures that the acceptor shutdown is not blocked waiting on the processors to be fully shutdown by decoupling the shutdown signal and the awaiting. It also ensure that the processor threads drain its newConnection queue to unblock acceptors that may be waiting. However, the acceptors still bind during the startup, only the processing of new connections and requests is further delayed.

The flow looks like this now:

val socketServer = ...

socketServer.startup(startProcessingRequests = false)
// Acceptors are bound.

socketServer.startProcessingRequests(authorizerFutures)
// Acceptors and Processors process new connections and requests

socketServer.stopProcessingRequests()
// Acceptors and Processors are stopped

socketServer.shutdown()
// SocketServer is shutdown (metrics, etc.)

Committer Checklist (excluded from commit message)

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

…itions

ijuma · 2020-04-08T13:49:51Z

always stop the processor threads before the acceptor threads.

I would have thought we want to stop accepting before we stop processing.

dajac · 2020-04-08T15:03:18Z

always stop the processor threads before the acceptor threads.

I would have thought we want to stop accepting before we stop processing.

Indeed, it is a bit counter intuitive. The reason behind is that the Acceptor can be blocked by the Processor and thus can't be shutdown when it happens. We could probably keep a more intuitive ordering by decoupling the shutdown and the awaiting of the shutdown. Let me check this.

dajac · 2020-04-09T08:16:12Z

@rajinisivaram The PR is ready to be reviewed.

rajinisivaram · 2020-04-14T14:24:53Z

ok to test

rajinisivaram

@dajac Thanks for the PR, looks good. Left some minor comments.

rajinisivaram · 2020-04-14T14:29:44Z

-   * is used to delay processing client connections until server is fully initialized, e.g.
-   * to ensure that all credentials have been loaded before authentications are performed.
-   * Acceptors are always started during `startup` so that the bound port is known when this
-   * method completes even when ephemeral ports are used. Incoming connections on this server


These two lines are still true, but removed from the comment?

Partially. The acceptors are not started but start to listen. Let me rework the comment to include the part about the bound port though.

rajinisivaram · 2020-04-14T14:29:45Z

-   * is used to delay processing client connections until server is fully initialized, e.g.
-   * to ensure that all credentials have been loaded before authentications are performed.
-   * Acceptors are always started during `startup` so that the bound port is known when this
-   * method completes even when ephemeral ports are used. Incoming connections on this server


These two lines are still true, but removed from the comment?

rajinisivaram · 2020-04-14T14:31:03Z

-   * This method is used for delayed starting of data-plane processors if [[kafka.network.SocketServer#startup]]
-   * was invoked with `startupProcessors=false`.
+   * Start processing requests and new connections. This method is used for delayed starting of
+   * data-plane processors if [[kafka.network.SocketServer#startup]] was invoked with


this is not just data-plane processors?

Correct. Let me rework the comment.

rajinisivaram · 2020-04-14T14:31:46Z

+   * listener before other listeners. This allows authorization metadata for other listeners to be
+   * stored in Kafka topics in this cluster.
+   *
+   * @param authorizerFutures


Add a description?

rajinisivaram · 2020-04-14T14:50:14Z

   */
  private def closeAll(): Unit = {
+    // Clear to unblock blocked acceptors
+    newConnections.asScala.foreach(_.close())


The blocked acceptor would then add another connection to this list right? Do we close that one?

No, we don't close that one. Let me rework this.

rajinisivaram · 2020-04-14T14:52:56Z

      externalReadyFuture.complete(null)
      TestUtils.waitUntilTrue(() => listenerStarted(externalListener), "External listener not started")
    } finally {
+      externalReadyFuture.complete(null)


Why? If it for the failure case, then perhaps it should be in a catch block?

This is not needed. It is a left over from my debugging. Let me remove it.

rajinisivaram · 2020-04-14T14:55:41Z

+    connect(testableServer, new ListenerName("EXTERNAL"), localAddr = InetAddress.getLocalHost)
+
+    // Wait to let the acceptor accepts the connections
+    Thread.sleep(100)


Can we replace sleep with some condition?

I have reworked this test. It still consistently fails without this patch and it does not have the sleep any more.

rajinisivaram · 2020-04-14T14:58:19Z

retest this please

dajac · 2020-04-15T08:47:03Z

@rajinisivaram Thanks for the review! I have addressed on your comments. Could you please have another look at it?

rajinisivaram

@dajac Thanks for the updates, looks good. Left just a couple of minor comments.

rajinisivaram · 2020-04-16T10:51:50Z

-   * Initiates a graceful shutdown by signaling to stop and waiting for the shutdown to complete
+   * Initiates a graceful shutdown by signaling to stop
   */
  def shutdown(): Unit = {


Should we rename this method to be initiateShutdown() to be consistent with kafka.utils.ShutdownableThread?

rajinisivaram · 2020-04-16T10:54:05Z

+    while (!newConnections.isEmpty) {
+      newConnections.poll().close()
+    }
+    newConnections.clear()


clear() is unnecessary since we would expect the loop to clear (i.e. we shouldn't have code that clears without closing).

rajinisivaram · 2020-04-16T10:55:16Z

retest this please

dajac · 2020-04-16T11:19:30Z

@rajinisivaram Thanks. I have addressed your comments.

rajinisivaram

@dajac Thanks for the updates, LGTM.

rajinisivaram · 2020-04-16T11:33:08Z

ok to test

rajinisivaram · 2020-04-16T12:36:25Z

retest this please

rajinisivaram · 2020-04-16T16:11:22Z

Test failure not related, merging to trunk.

* 'trunk' of github.com:apache/kafka: (28 commits) MINOR: cleanup RocksDBStore tests (apache#8510) KAFKA-9818: Fix flaky test in RecordCollectorTest (apache#8507) MINOR: reduce impact of trace logging in replica hot path (apache#8468) KAFKA-6145: KIP-441: Add test scenarios to ensure rebalance convergence (apache#8475) KAFKA-9881: Convert integration test to verify measurements from RocksDB to unit test (apache#8501) MINOR: improve test coverage for dynamic LogConfig(s) (apache#7616) MINOR: Switch order of sections on tumbling and hopping windows in streams doc. Tumbling windows are defined as "special case of hopping time windows" - but hopping windows currently only explained later in the docs. (apache#8505) KAFKA-9819: Fix flaky test in StoreChangelogReaderTest (apache#8488) HOTFIX: fix active task process ratio metric recording KAFKA-9796; Ensure broker shutdown is not stuck when Acceptor is waiting on connection queue (apache#8448) MINOR: Use streaming iterator with decompression buffer when building offset map (apache#8494) Add log message in release.py (apache#8461) KAFKA-9854 Re-authenticating causes mismatched parse of response (apache#8471) KAFKA-9838; Add log concurrency test and fix minor race condition (apache#8476) KAFKA-9703; Free up compression buffer after splitting a large batch KAFKA-9779: Add Stream system test for 2.5 release (apache#8378) KAFKA-7885: TopologyDescription violates equals-hashCode contract. (apache#6210) MINOR: KafkaApis#handleOffsetDeleteRequest does not group result correctly (apache#8485) HOTFIX: don't close or wipe out someone else's state (apache#8478) MINOR: add process(Test)Messages to the README (apache#8480) ...

KAFKA-9796; Broker shutdown could be stuck forever under certain cond…

e347ce3

…itions

ijuma requested a review from rajinisivaram April 8, 2020 13:47

fixup

494e958

dajac added 2 commits April 8, 2020 17:21

keep a more intuitive shutdown ordering: 1) acceptor; 2) processors

0f8a9dc

fixup

5565c89

dajac changed the title ~~[WIP] KAFKA-9796; Broker shutdown could be stuck forever under certain conditions~~ KAFKA-9796; Broker shutdown could be stuck forever under certain conditions Apr 9, 2020

rajinisivaram reviewed Apr 14, 2020

View reviewed changes

Address review

51925b7

rajinisivaram reviewed Apr 16, 2020

View reviewed changes

Address comments

7c8251a

rajinisivaram approved these changes Apr 16, 2020

View reviewed changes

rajinisivaram merged commit 9a36d9f into apache:trunk Apr 16, 2020

dajac deleted the KAFKA-9796 branch April 16, 2020 16:16

Conversation

dajac commented Apr 8, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

ijuma commented Apr 8, 2020

Uh oh!

dajac commented Apr 8, 2020

Uh oh!

dajac commented Apr 9, 2020

Uh oh!

rajinisivaram commented Apr 14, 2020

Uh oh!

rajinisivaram 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!

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!

rajinisivaram commented Apr 14, 2020

Uh oh!

dajac commented Apr 15, 2020

Uh oh!

rajinisivaram 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!

rajinisivaram commented Apr 16, 2020

Uh oh!

dajac commented Apr 16, 2020

Uh oh!

rajinisivaram left a comment

Choose a reason for hiding this comment

Uh oh!

rajinisivaram commented Apr 16, 2020

Uh oh!

rajinisivaram commented Apr 16, 2020

Uh oh!

rajinisivaram commented Apr 16, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

dajac commented Apr 8, 2020 •

edited

Loading