| 503 | |
| 504 | |
| 505 | void GroupProcess::expired(int64_t sessionId) |
| 506 | { |
| 507 | if (error.isSome() || sessionId != zk->getSessionId()) { |
| 508 | return; |
| 509 | } |
| 510 | |
| 511 | LOG(INFO) << "ZooKeeper session expired"; |
| 512 | |
| 513 | // Cancel the retries. Group will sync() after it reconnects to ZK. |
| 514 | retrying = false; |
| 515 | |
| 516 | // Cancel and cleanup the connect timer (if necessary). |
| 517 | if (connectTimer.isSome()) { |
| 518 | Clock::cancel(connectTimer.get()); |
| 519 | connectTimer = None(); |
| 520 | } |
| 521 | |
| 522 | // From the group's local perspective all the memberships are |
| 523 | // gone so we need to update the watches. |
| 524 | // If the memberships still exist on ZooKeeper, they will be |
| 525 | // restored in group after the group reconnects to ZK. |
| 526 | // This is a precaution against the possibility that ZK connection |
| 527 | // is lost right after we recreate the ZK instance below or the |
| 528 | // entire ZK cluster goes down. The outage can last for a long time |
| 529 | // but the clients watching the group should be informed sooner. |
| 530 | memberships = set<Group::Membership>(); |
| 531 | update(); |
| 532 | |
| 533 | // Invalidate the cache so that we'll sync with ZK after |
| 534 | // reconnection. |
| 535 | memberships = None(); |
| 536 | |
| 537 | // Set all owned memberships as cancelled. |
| 538 | foreachpair (int32_t sequence, Promise<bool>* cancelled, utils::copy(owned)) { |
| 539 | cancelled->set(false); // Since this was not requested. |
| 540 | owned.erase(sequence); // Okay since iterating over a copy. |
| 541 | delete cancelled; |
| 542 | } |
| 543 | |
| 544 | CHECK(owned.empty()); |
| 545 | |
| 546 | // Note that we DO NOT clear unowned. The next time we try and cache |
| 547 | // the memberships we'll trigger any cancelled unowned memberships |
| 548 | // then. We could imagine doing this for owned memberships too, but |
| 549 | // for now we proactively cancel them above. |
| 550 | |
| 551 | state = DISCONNECTED; |
| 552 | |
| 553 | delete CHECK_NOTNULL(zk); |
| 554 | delete CHECK_NOTNULL(watcher); |
| 555 | startConnection(); |
| 556 | } |
| 557 | |
| 558 | |
| 559 | void GroupProcess::updated(int64_t sessionId, const string& path) |