| 333 | |
| 334 | |
| 335 | void GroupProcess::connected(int64_t sessionId, bool reconnect) |
| 336 | { |
| 337 | if (error.isSome() || sessionId != zk->getSessionId()) { |
| 338 | return; |
| 339 | } |
| 340 | |
| 341 | LOG(INFO) << "Group process (" << self() << ") " |
| 342 | << (reconnect ? "reconnected" : "connected") << " to ZooKeeper"; |
| 343 | |
| 344 | if (!reconnect) { |
| 345 | // This is the first time the ZooKeeper client connects to |
| 346 | // ZooKeeper service. (It could be also the first time for the |
| 347 | // group or after session expiration which causes a new ZooKeeper |
| 348 | // client instance to be created.) |
| 349 | CHECK_EQ(state, CONNECTING); |
| 350 | state = CONNECTED; |
| 351 | } else { |
| 352 | // This means we are reconnecting within the same ZooKeeper |
| 353 | // session. We could have completed authenticate() or create() |
| 354 | // before we lost the connection (thus the state can be the any |
| 355 | // of the following three) so 'sync()' below will check the state |
| 356 | // and only execute necessary operations accordingly. |
| 357 | CHECK(state == CONNECTED || state == AUTHENTICATED || state == READY) |
| 358 | << state; |
| 359 | } |
| 360 | |
| 361 | // Cancel and cleanup the connect timer. The timer should always be |
| 362 | // set, because it is set before making the initial connection |
| 363 | // attempt and whenever a reconnection attempt is made. |
| 364 | CHECK_SOME(connectTimer); |
| 365 | |
| 366 | // Now that we are connected, we'll learn about a subsequent |
| 367 | // disconnection event via the `reconnecting` callback. At that |
| 368 | // point we'll also restart the `connectTimer` to ensure we retry |
| 369 | // the reconnection attempt. |
| 370 | Clock::cancel(connectTimer.get()); |
| 371 | connectTimer = None(); |
| 372 | |
| 373 | // Sync group operations (and set up the group on ZK). |
| 374 | Try<bool> synced = sync(); |
| 375 | |
| 376 | if (synced.isError()) { |
| 377 | // Non-retryable error. Abort. |
| 378 | abort(synced.error()); |
| 379 | } else if (!synced.get()) { |
| 380 | // Retryable error. |
| 381 | if (!retrying) { |
| 382 | delay(RETRY_INTERVAL, self(), &GroupProcess::retry, RETRY_INTERVAL); |
| 383 | retrying = true; |
| 384 | } |
| 385 | } |
| 386 | } |
| 387 | |
| 388 | |
| 389 | Try<bool> GroupProcess::authenticate() |