KAFKA-7165: Retry the BrokerInfo registration into ZooKeeper #5575

jonathansantilli · 2018-08-26T19:46:50Z

The following is a proposal when the ZooKeeper session has changed and the Broker tries to register himself into ZooKeeper.

Currently, this throws a NodeExistsException since a Broker with the same id got registered previously into Zookeeper.

Assuming Broker id = 1

If a Broker using this patch retries to register himself into ZooKeeper and the previous ZooKeeper session was not the one that registers the Broker (another Broker with the same id did), the Broker won't be allowed to register and a NodeExistsException will be generated.
If a Broker using this patch retries to register himself into ZooKeeper and the previous ZooKeeper session was the one that registers the Broker, the ephemeral node will be deleted from ZooKeeper and re-created by the Broker.
If a Broker not using this patch retries to register himself into ZooKeeper and the previous ZooKeeper session was not the one that registers the Broker (another Broker with the same id did), the Broker won't be allowed to register and a NodeExistsException will be generated.

Committer Checklist (excluded from commit message)

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

junrao · 2018-08-28T01:17:40Z

@jonathansantilli : Thanks for the patch. About the approach. I am not exactly sure about the impact of explicitly deleting an ephemeral node whose session is about to expire. For example, I am not sure if the following can happen: (1) ZK session 1 expires and the ZK leader changes; (2) new ZK leader retains the ephemeral node from session 1 and is about to expire session 1; (3) ZK session 2 deletes and recreates the ephemeral node; (4) ZK leader expires session 1 and deletes the ephemeral node created by session 2. If this can happen, it will create a weird problem in Kafka.

An alternative approach is to retry the creation of the ephemeral node up to sth like twice the session timeout. It may take a bit long for the broker to be re-registered. However, it seems it's a bit safer and simpler, until ZOOKEEPER-2985 is fixed.

jonathansantilli · 2018-09-12T07:01:34Z

Hello @junrao thanks a lot for your comment.
Am pretty sure you know way better than me how Kafka + Zookeeper interaction works.
I just want to point out that ephemeral node gets created immediately after deleting the one created by the previous session (which belongs to the Broker who creates it).

Do you mind If I move the conversation to the JIRA issue and there we can elaborate your alternative approach? I would like to continue working on this.

jonathansantilli · 2018-10-24T08:35:06Z

Hello @junrao, I would like to continue working on this bug, hope you can have some time to elaborate a little bit more your proposal here: https://issues.apache.org/jira/browse/KAFKA-7165?focusedCommentId=16661931&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#comment-16661931

Cheers!

junrao

@jonathansantilli : On second thought, your current patch could also work as a temporary fix. Added a few more comments below.

junrao · 2018-10-29T23:44:09Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

+  * is created and the subsequent updates of the session id, it is possible that the session id change
+  * over the time for 'Session expired', for instance.
+  * */
+  private var previousZooKeeperSessionId = zooKeeperClient.sessionId


The following is the comment from ZooKeeper.getSessionId(). Since ZK connects to the server asynchronously, the sessionId is not guaranteed to be available until the session is established.

/** * The session id for this ZooKeeper client instance. The value returned is * not valid until the client connects to a server and may change after a * re-connect. * * This method is NOT thread safe * * @return current session id */ public long getSessionId() {

Oh, this could be a problem at the moment the session Id is collected for the first time. I will check where is the best place to assign it.
Thanks for the heads up.

junrao · 2018-10-29T23:49:59Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

      getDataResponse.resultCode match {
-        case Code.OK if getDataResponse.stat.getEphemeralOwner != zooKeeperClient.sessionId =>
+        case Code.OK if shouldReCreateEphemeralZNode(ephemeralOwnerId) =>


Could we add some comments why we need this and that we want to remove this once ZOOKEEPER-2985 is fixed?

Yes, is definitely needed, I will add a comment with the appropriated references.
Once again, thanks for the heads up.

junrao · 2018-10-29T23:51:19Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

+  * */
+  private var previousZooKeeperSessionId = zooKeeperClient.sessionId
+  // Only for testing, to simulate the ZooKeeper session id has changed
+  private var currentZooKeeperSessionId: Long = -1


Do we really need this? Could we just create a subclass of KafkaZkClient and overwrite the return value of shouldReCreateEphemeralZNode() in the test?

As you pointed out, this is a momentaneous workaround to tackle the issue. I have thought on a more elegant way to do it, as you propose. Probably am opting for the simples but not elegant solution for this. The rationale behind my proposal is:

Keeping everything in the same place/file (now with the comments you suggested is better), in this way, the workaround is framed to a single place.

Once the Zookeeper issue is addressed properly, we just have to come to this file/class and delete this code.

Having said that, am more than happy to change it if you think is better :)

Yes, I am mostly after simplicity. The patch as it is seems a bit hard for me to read. Part of that is to have to understand the meaning of 2 separate ZK session ids and when they are updated. I thought maintaining just one session id var could be simpler. With regard to keeping things in one place, I guess in either case, we have to change the test case when this logic is eventually removed?

Yes, it makes sense, am changing the code to achieve that. 👍

jonathansantilli · 2018-11-02T12:04:24Z

Hello @junrao I have done the suggested changes, please take a look when suitable for you 😄

junrao

@jonathansantilli : Thanks for the updated patch. A few more comments below.

junrao · 2018-11-02T23:42:08Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

@@ -67,6 +67,12 @@ class KafkaZkClient private (zooKeeperClient: ZooKeeperClient, isSecure: Boolean
  // Only for testing
  private[kafka] def currentZooKeeper: ZooKeeper = zooKeeperClient.currentZooKeeper

+  // This variable holds the Zookeeper session id at the moment a Broker gets registered in Zookeeper and the subsequent
+  // updates of the session id. It is possible that the session id change over the time for 'Session expired'.


change -> changes

junrao · 2018-11-02T23:43:46Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

+  // This variable holds the Zookeeper session id at the moment a Broker gets registered in Zookeeper and the subsequent
+  // updates of the session id. It is possible that the session id change over the time for 'Session expired'.
+  // This code is part of the work around done in the KAFKA-7165, once ZOOKEEPER-2985 is complete, this code must
+  // be deleted, if by the time you are reading this the code is still present and the bug solve, please delete it.


We probably don't need the remaining part starting with "if by the time". Ditto below.

junrao · 2018-11-02T23:59:30Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

-        case Code.OK if getDataResponse.stat.getEphemeralOwner != zooKeeperClient.sessionId =>
+        // At this point, the Zookeeper session could be different (due a 'Session expired') from the one that initially
+        // registered the Broker into the Zookeeper ephemeral node, but the znode is still present in ZooKeeper.
+        // The expected behaviour is that Zookeeper removes the ephemeral node associated with the expired session but


Zookeeper removes => Zookeeper server removes

junrao · 2018-11-03T00:06:52Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

+      // That is, the fist attempt to register the Broker, we save a reference of the used Zookeeper session id.
+      // This is done here since the Zookeeper session id may not be available at the Object creation time,
+      // hence, this is assuming the 'retryRequestUntilConnected' method got connected and a valid session id is present.
+      setCurrentZookeeperSessionId()


Hmm, do we need both setCurrentZookeeperSessionId() and updateCurrentZKSessionId()? To me, it seems that we can just initialize currentZooKeeperSessionId to -1 and set currentZooKeeperSessionId
in checkedEphemeralCreate(path: String, data: Array[Byte]) if the return code is ok.

junrao · 2018-11-03T00:09:19Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

+      val deleteResponse = retryRequestUntilConnected(deleteRequest)
+      deleteResponse.resultCode match {
+        case code@ Code.OK => code
+        case code =>


It seems that if code is NoNode, we should just let it go.

@junrao sorry, but I do not get this comment.
Is true (after you point this out) that at the moment we try to delete the ephemeral znode, it could not be present anymore an a NoNode response will be received, and because we are just checking if (codeAfterDelete == Code.OK) the created() call will not be performed, hence, the Broker will be not registered. Is that what you meant?

If so, will be adding if (codeAfterDelete == Code.OK || codeAfterDelete == Code.NONODE)?

Hope you can help me to understand your point.

Cheers!

Or this?

deleteResponse.resultCode match { case code@ Code.OK => code > case code@ Code.NONODE => code case code => error(s"Error while deleting ephemeral node at $path with return code: $code") code }

Yes, it seems that if the node doesn't exist (ZK server could have deleted), we should still do the create() instead of returning an error code.

junrao · 2018-11-03T00:10:54Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

+    private def reCreate(): Code = {
+      val codeAfterDelete = delete()
+      var codeAfterReCreate = codeAfterDelete
+      info(s"Result of znode ephemeral deletion at $path is: $codeAfterDelete")


It seems this should be debug level logging?

junrao · 2018-11-03T00:10:59Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

+      info(s"Result of znode ephemeral deletion at $path is: $codeAfterDelete")
+      if (codeAfterDelete == Code.OK) {
+        codeAfterReCreate = create()
+        info(s"Result of znode ephemeral re-creation at $path is: $codeAfterReCreate")


It seems this should be debug level logging?

junrao · 2018-11-03T00:12:15Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

+        // This code is part of the work around done in the KAFKA-7165, once ZOOKEEPER-2985 is complete, this code must
+        // be deleted, if by the time you are reading this the code is still present and the bug solve, please delete it.
+        case Code.OK if shouldReCreateEphemeralZNode(ephemeralOwnerId) =>
+          warn(s"Error while creating ephemeral at $path, node already exists and owner " +


Should this be info instead of warn? We probably want to void the word Error in an info logging.

jonathansantilli · 2018-11-03T23:38:29Z

Hello @junrao I have done the suggested changes, please take a look. Cheers!

junrao

@jonathansantilli : Thanks for the updated patch. Just a couple of more comments below.

junrao · 2018-11-05T21:27:02Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

+      val deleteRequest = DeleteRequest(path, ZkVersion.MatchAnyVersion)
+      val deleteResponse = retryRequestUntilConnected(deleteRequest)
+      deleteResponse.resultCode match {
+        case code@ Code.OK => code


Could we just move the call to updateCurrentZKSessionId() here? This way, there will be a single place where updateCurrentZKSessionId() will be called.

Hmmmm, where exactly @junrao?

We are setting the currentZooKeeperSessionId at the end of private def checkedEphemeralCreate(path: String, data: Array[Byte]) method, this ensures that the first time the broker tries to registered we have the session Id used.

If we do not do that, the method isZKSessionTheEphemeralOwner will return false (since -1 != zooKeeperClient.sessionId) and will not try to recreate the ephemeral node.

Then we are setting currentZooKeeperSessionId again if the condition case Code.OK if shouldReCreateEphemeralZNode(ephemeralOwnerId) => applies, becase this will update the currentZooKeeperSessionId with the newest session, useful when trying to recreate the ephemeral node.

Am pretty sure am missing something since you point this out, but I can not see it, hope you can give me a clue :)

Sorry, I didn't make it clear. Basically, we want to set the session id after a successful creation of the ephemeral path. I was thinking that we can call updateCurrentZKSessionId() after line 1667 in CheckedEphemeral.create() when the return code is ok. Then, we don't need to call updateCurrentZKSessionId() anywhere else.

junrao · 2018-11-05T21:32:08Z

core/src/test/scala/unit/kafka/zk/KafkaZkClientTest.scala

+    expiredSessionZkClient.registerBroker(expiredBrokerInfo)
+
+    // The broker info should be the same, no error should be raised
+    assertEquals(Some(expiredBrokerInfo.broker), expiredSessionZkClient.getBroker(brokerId))


Could we further verify that the czxid of the ephemeral node is different?

Yes, thanks, good idea, doing it.

jonathansantilli · 2018-11-06T21:56:54Z

Hello @junrao, I have done the suggested changes, please take a look. Cheers!

junrao

@jonathansantilli : Thanks for the new update. A couple of more comments below.

junrao · 2018-11-06T23:10:27Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

+          // This is assuming the 'retryRequestUntilConnected' method got connected and a valid session id is present.
+          // This code is part of the work around done in the KAFKA-7165, once ZOOKEEPER-2985 is complete, this code
+          // must be deleted.
+          updateCurrentZKSessionId(zooKeeperClient.sessionId)


Actually, we need to update the ZK session id in the case below when getAfterNodeExists(0 returns true. So, perhaps we can save the code for all cases and call updateCurrentZKSessionId() if the saved code is OK.

I think you mean when getAfterNodeExists() returns Code.OK.
If I understood correctly, will be like:

var createResultCode = Code.OK createResponse.resultCode match { case code@ Code.OK => createResultCode = code case Code.NODEEXISTS => createResultCode = getAfterNodeExists() case code => error(s"Error while creating ephemeral at $path with return code: $code") createResultCode = code } if (createResultCode == Code.OK) { // At this point, we need to save a reference to the zookeeper session id. // This is done here since the Zookeeper session id may not be available at the Object creation time. // This is assuming the 'retryRequestUntilConnected' method got connected and a valid session id is present. // This code is part of the workaround done in the KAFKA-7165, once ZOOKEEPER-2985 is complete, this code // must be deleted. updateCurrentZKSessionId(zooKeeperClient.sessionId) } createResultCode

What do you think?

Yes, that's exactly what I was thinking.

Cool @junrao, I have updated the code as suggested.

junrao · 2018-11-06T23:13:04Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

@@ -1585,7 +1591,7 @@ class KafkaZkClient private (zooKeeperClient: ZooKeeperClient, isSecure: Boolean

  private def acls(path: String): Seq[ACL] = ZkData.defaultAcls(isSecure, path)

-  private def retryRequestUntilConnected[Req <: AsyncRequest](request: Req): Req#Response = {
+  private[zk] def retryRequestUntilConnected[Req <: AsyncRequest](request: Req): Req#Response = {


Do we need to expose package level visibility for this method?

Is to take advantage of that method to perform the call to Zookeeper.

I can move the method getPathCzxid present in the class ExpiredKafkaZkClient from the test, to the class KafkaZkClient (but not useful in that class KafkaZkClient).

Or I can add more code just to make the call to Zookeeper to get the czxid in the class ExpiredKafkaZkClient

What do you think?

junrao

@jonathansantilli : Thanks for the new patch. LGTM. Just a few more minor comments below.

junrao · 2018-11-08T00:37:59Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

        case code =>
          error(s"Error while creating ephemeral at $path with return code: $code")
+          createResultCode = code
+      }


Perhaps the code can be simplified a bit as the following?

val createResultCode = createResponse.resultCode match { case code@ Code.OK => code case Code.NODEEXISTS => getAfterNodeExists() case code => error(s"Error while creating ephemeral at $path with return code: $code") code }

Yes, it definitely is simpler and allow use to use val instead of var.
Doing it.

junrao · 2018-11-08T00:42:33Z

core/src/test/scala/unit/kafka/zk/KafkaZkClientTest.scala

@@ -58,21 +58,26 @@ class KafkaZkClientTest extends ZooKeeperTestHarness {
  val topicPartitions10_11 = Seq(topicPartition10, topicPartition11)
  val controllerEpochZkVersion = 0

-  var otherZkClient: KafkaZkClient = _
+  var otherZkClient : KafkaZkClient = _
+  var expiredSessionZkClient : ExpiredKafkaZkClient = _


The convention seems to be no space before ":" in both line 61 and 62.

junrao · 2018-11-08T00:44:30Z

core/src/test/scala/unit/kafka/zk/KafkaZkClientTest.scala

+    override def shouldReCreateEphemeralZNode(ephemeralOwnerId: Long): Boolean = {
+      true
+    }
+    def getPathCzxid(path: String): Long = {


Could we add a new line above?

In case the ZooKeeper session has been regenerated and the broker tries to register the BrokerInfo into Zookeeper, this code deletes the current BrokerInfo from Zookeeper and creates it again, just if the znode ephemeral owner belongs to the Broker which tries to register himself again into ZooKeeper

jonathansantilli · 2018-11-08T18:15:05Z

Hello @junrao, I have added the changes as suggested.

jonathansantilli · 2018-11-08T19:03:03Z

@junrao please let me know what's the process to merge this PR, please :)

junrao

@jonathansantilli : Thanks for the latest patch. LGTM

pabloa502 · 2019-01-03T13:33:44Z

Good day all, This is not yet included in any release, right? How to work around it?

jonathansantilli · 2019-01-05T02:16:20Z

Hello @pabloa502 although this has been included into the trunk branch, is not yet released.
Here are the release notes of the current stable version (at the moment am writing this) https://www.apache.org/dist/kafka/2.1.0/RELEASE_NOTES.html

…per (apache#5575) * Add logic to retry the BrokerInfo registration into ZooKeeper In case the ZooKeeper session has been regenerated and the broker tries to register the BrokerInfo into Zookeeper, this code deletes the current BrokerInfo from Zookeeper and creates it again, just if the znode ephemeral owner belongs to the Broker which tries to register himself again into ZooKeeper * Add test to validate the BrokerInfo re-registration into ZooKeeper

…5575) * Add logic to retry the BrokerInfo registration into ZooKeeper In case the ZooKeeper session has been regenerated and the broker tries to register the BrokerInfo into Zookeeper, this code deletes the current BrokerInfo from Zookeeper and creates it again, just if the znode ephemeral owner belongs to the Broker which tries to register himself again into ZooKeeper * Add test to validate the BrokerInfo re-registration into ZooKeeper

ijuma requested a review from junrao August 27, 2018 02:20

jonathansantilli force-pushed the KAFKA-7165 branch from 22fc1a0 to 5ffc8a7 Compare October 29, 2018 19:02

junrao reviewed Oct 29, 2018

View reviewed changes

jonathansantilli force-pushed the KAFKA-7165 branch from 5ffc8a7 to 15c6795 Compare November 2, 2018 11:04

junrao reviewed Nov 3, 2018

View reviewed changes

jonathansantilli force-pushed the KAFKA-7165 branch from 15c6795 to 2b30a71 Compare November 3, 2018 23:33

junrao reviewed Nov 5, 2018

View reviewed changes

jonathansantilli force-pushed the KAFKA-7165 branch from 2b30a71 to f7f0cb1 Compare November 6, 2018 21:54

junrao reviewed Nov 6, 2018

View reviewed changes

jonathansantilli force-pushed the KAFKA-7165 branch from f7f0cb1 to 191d5ae Compare November 7, 2018 22:10

junrao approved these changes Nov 8, 2018

View reviewed changes

Jonathan Santilli added 2 commits November 8, 2018 18:12

Add test to validate the BrokerInfo re-registration into ZooKeeper

b0e2df5

jonathansantilli force-pushed the KAFKA-7165 branch from 191d5ae to b0e2df5 Compare November 8, 2018 18:13

junrao approved these changes Nov 8, 2018

View reviewed changes

junrao merged commit eb3335e into apache:trunk Nov 8, 2018

KAFKA-7165: Retry the BrokerInfo registration into ZooKeeper #5575

KAFKA-7165: Retry the BrokerInfo registration into ZooKeeper #5575

Conversation

jonathansantilli commented Aug 26, 2018

Committer Checklist (excluded from commit message)

junrao commented Aug 28, 2018

jonathansantilli commented Sep 12, 2018

jonathansantilli commented Oct 24, 2018

junrao left a 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

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

jonathansantilli commented Nov 2, 2018

junrao left a 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

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

Choose a reason for hiding this comment

jonathansantilli commented Nov 3, 2018

junrao left a 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jonathansantilli commented Nov 6, 2018

junrao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jonathansantilli Nov 7, 2018 • edited Loading

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

jonathansantilli Nov 7, 2018 • edited Loading

Choose a reason for hiding this comment

junrao left a 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

jonathansantilli commented Nov 8, 2018

jonathansantilli commented Nov 8, 2018

junrao left a comment

Choose a reason for hiding this comment

pabloa502 commented Jan 3, 2019

jonathansantilli commented Jan 5, 2019

jonathansantilli Nov 7, 2018 •

edited

Loading

jonathansantilli Nov 7, 2018 •

edited

Loading