5 ms·

In the days before hosted Kafka implementations were readily available, I tried to set up Kafka in our AWS infrastructure. Getting Zookeeper working in a world

by kevsim 7y ago

In the days before hosted Kafka implementations were readily available, I tried to set up Kafka in our AWS infrastructure. Getting Zookeeper working in a world based on autoscaling groups was a nightmare. It felt like it was built for the days where each server was a special snowflake. Looking forward to seeing if this gains traction.

rad_gruchalski 7y ago
This is just an FYI type of comment. with zookeeper, one needs an odd number of machines, usually not greater than seven. Servers are indexed from 1 to n. They’re replaced one for one, so the server id matches. Scaling it up and down is a bit of a pita because every server needs to know all other servers. Changing the ensemble members requires restarting every zookeeper instance because zookeeper doesn’t support config reload. Zookeeper does expect a single “host” for every server but it runs perfectly fine with docker. But that’s no different from consul or etcd.
- vetrom 7y ago
  Zookeeper has implemented dynamic reconfiguration since circa 2013-2015ish, but it required support for the feature in clients, as well as servers. That said, implementing an autoscaler that manages that as well was not trivial, in my experience.
  - lars_francke 7y ago
    This is only half true. Yes, the dynamic reconfiguration features has been in trunk for a while but ZooKeeper 3.5 which is the first GA release to include this has only been released 4-8 weeks ago.
- thomaslee 7y ago
  Another FYI type comment I guess. :) Some of my more gripey statements here may be outdated info, so DYOR I guess. Dynamic reconfig in 3.5 addresses the "restarting every zookeeper instance" problem. [0] You stand up an initial quorum with seed config, then tie in new servers with "reconfig -add". Not sure how well it would tie into cloudy autoscaling stuff though. I wouldn't start there myself. A much bigger pain IMO is the handling of DNS in the official Java ZK client earlier than 3.4.13/3.5.5 (and by association, Curator, ZkClient, etc.). [1] The former was released mid 2018 and the latter this year, so tons of stuff out there that just won't find a host if IPs change. If you "own" all the clients it's maybe not a problem, but if you've got a lot of services owned by a ton of teams it's ... challenging. Even with the fix for ZOOKEEPER-2184 in place I'm pretty sure DNS lookups are only retried if a connect fails, so there's still the issue of IPs "swapping" unexpectedly at the wrong time in cloud environments which can lead to a ZK server in cluster A talking to a ZK server in cluster B (or worse: clients of cluster A talking to cluster B mistakenly thinking that they're talking to cluster A). I'm sure this problem's not unique to ZK though. Authentication helps prevent the worst-case scenarios, but I'm not sure if it helps from an uptime perspective. TL;DR: ZK in the cloud can get messy (even if you play it relatively "safe"). [0] https://zookeeper.apache.org/doc/r3.5.5/zookeeperReconfig.html https://zookeeper.apache.org/doc/r3.5.5/zookeeperReconfig.ht... [1] https://issues.apache.org/jira/browse/ZOOKEEPER-2184 https://issues.apache.org/jira/browse/ZOOKEEPER-2184
  - rad_gruchalski 7y ago
    Good to know! Thank you! Haven’t dealt with zk in detail since ~2016.
  - deleted 7y ago
    [deleted]
- hinkley 7y ago
  There are some other consensus domains that have trouble when members disappear or arrive permanently, and this has kinda been kicking around my head for a while: Should it not be the case that one of the responsibilities of the quorum is to vote new members in or out? I mean, as a first-class feature of the system. The main problem I see is that in most consensus systems, any member can be nominated as leader. But if the member was inducted during a partition event - which one could do if the partition were long duration - then nominations for new leaders will go out. What happens if the new member gets elected? How do the partitioned machines find that leader when they return? So the process of joining the quorum would have to be incremental. Because demanding a unanimous vote to add a machine means you can never replace a dead one (except by impersonating it)
  - benesch 7y ago
    > Should it not be the case that one of the responsibilities of the quorum is to vote new members in or out? I mean, as a first-class feature of the system. Indeed. What you're describing is one of the main motivations for Raft. Paxos showed that distributed consensus was mathematically sound, but did little to guide implementors in actually building such a system. Raft is not a fundamentally new consensus algorithm; just an incremental improvement that formalizes many of the improvements that you needed to make to Paxos anyway, like membership changes, log compaction, and multi-decree support from the get go. If you're interested, the Raft paper is quite readable and goes into this in detail. [0] > The main problem I see is that in most consensus systems, any member can be nominated as leader. But if the member was inducted during a partition event - which one could do if the partition were long duration - then nominations for new leaders will go out. What happens if the new member gets elected? How do the partitioned machines find that leader when they return? How do the partitioned machines find that leader when they return? This isn't actually the tricky bit, as it turns out. Communication flows from the leader to the other nodes, so the leader, even if it's a newly-inducted node, will initiate the connections to the partitioned nodes when the partition resolves. (The leader necessarily knows the addresses of the partitioned nodes, because the leader knows about all committed entries, and the identities of all the nodes in the cluster are committed into the Raft log.) Again, the Raft paper does a great job explaining cluster mebership changes—much better than I can! [0]: https://raft.github.io/raft.pdf https://raft.github.io/raft.pdf
WaxProlix 7y ago
Relieving to hear this as I've had almost exactly the same experience and it made me feel like an actual idiot.
necubi 7y ago
Using ASGs with zookeeper is kind of a pain, but also not very necessary. Production zookeeper clusters are almost always either 3, 5, or 7 nodes and it's very rare to change that once a cluster is in production. Given that this is all very static, the easiest thing is to just use terraform or similar to create the ec2 instances. If a node dies, you re-run your terraform to create a replacement. Alternatively, you can create an ASG-per-node so that you get auto-replacement. In my experience, ZK is one of the easiest and most reliable distributed systems to operate. I've only seen issues when it's used as a database instead of a distributed coordination service.
- coredog64 7y ago
  We ran into additional troubles based on ZK client bugs in the version that Kafka used. It would only ever resolve the ZK hostname once, at startup. Our workaround required some additional work to allocate an EIP for each host (3 AZs -> 3 EIPs) and then have the ZK host grab the EIP on startup. Even though Netflix hasn't updated it in a while, Exhibitor was helpful as well, in that it allowed ZK to bootstrap nodes off of state stored in S3. That did come at the cost of an extra 2-3 minutes per node on initial quorum startup.
  - uberduper 7y ago
    Might just need to set networkaddress.cache.ttl=60 in java.security.
fizx 7y ago
ASGs + Zookeeper is really easy, but you have to use https://github.com/soabase/exhibitor https://github.com/soabase/exhibitor. It coordinates replacing downed ZK nodes using S3, and also provides a bunch of extra management tools. I had a single Zookeeper ASG running under load for over three years without maintenance. I pinged my old co-founder to see if he's willing to open source the CloudFormation template.
RantyDave 7y ago
Yes, it was. More specifically Zookeeper is designed as a piece of infrastructure you put together then rely on.

by kevsim 7y ago

rad_gruchalski 7y ago

This is just an FYI type of comment. with zookeeper, one needs an odd number of machines, usually not greater than seven. Servers are indexed from 1 to n. They’re replaced one for one, so the server id matches. Scaling it up and down is a bit of a pita because every server needs to know all other servers. Changing the ensemble members requires restarting every zookeeper instance because zookeeper doesn’t support config reload. Zookeeper does expect a single “host” for every server but it runs perfectly fine with docker. But that’s no different from consul or etcd.

vetrom 7y ago
Zookeeper has implemented dynamic reconfiguration since circa 2013-2015ish, but it required support for the feature in clients, as well as servers. That said, implementing an autoscaler that manages that as well was not trivial, in my experience.
- lars_francke 7y ago
  This is only half true. Yes, the dynamic reconfiguration features has been in trunk for a while but ZooKeeper 3.5 which is the first GA release to include this has only been released 4-8 weeks ago.
thomaslee 7y ago
Another FYI type comment I guess. :) Some of my more gripey statements here may be outdated info, so DYOR I guess. Dynamic reconfig in 3.5 addresses the "restarting every zookeeper instance" problem. [0] You stand up an initial quorum with seed config, then tie in new servers with "reconfig -add". Not sure how well it would tie into cloudy autoscaling stuff though. I wouldn't start there myself. A much bigger pain IMO is the handling of DNS in the official Java ZK client earlier than 3.4.13/3.5.5 (and by association, Curator, ZkClient, etc.). [1] The former was released mid 2018 and the latter this year, so tons of stuff out there that just won't find a host if IPs change. If you "own" all the clients it's maybe not a problem, but if you've got a lot of services owned by a ton of teams it's ... challenging. Even with the fix for ZOOKEEPER-2184 in place I'm pretty sure DNS lookups are only retried if a connect fails, so there's still the issue of IPs "swapping" unexpectedly at the wrong time in cloud environments which can lead to a ZK server in cluster A talking to a ZK server in cluster B (or worse: clients of cluster A talking to cluster B mistakenly thinking that they're talking to cluster A). I'm sure this problem's not unique to ZK though. Authentication helps prevent the worst-case scenarios, but I'm not sure if it helps from an uptime perspective. TL;DR: ZK in the cloud can get messy (even if you play it relatively "safe"). [0] https://zookeeper.apache.org/doc/r3.5.5/zookeeperReconfig.html https://zookeeper.apache.org/doc/r3.5.5/zookeeperReconfig.ht... [1] https://issues.apache.org/jira/browse/ZOOKEEPER-2184 https://issues.apache.org/jira/browse/ZOOKEEPER-2184
- rad_gruchalski 7y ago
  Good to know! Thank you! Haven’t dealt with zk in detail since ~2016.
- deleted 7y ago
  [deleted]
hinkley 7y ago
There are some other consensus domains that have trouble when members disappear or arrive permanently, and this has kinda been kicking around my head for a while: Should it not be the case that one of the responsibilities of the quorum is to vote new members in or out? I mean, as a first-class feature of the system. The main problem I see is that in most consensus systems, any member can be nominated as leader. But if the member was inducted during a partition event - which one could do if the partition were long duration - then nominations for new leaders will go out. What happens if the new member gets elected? How do the partitioned machines find that leader when they return? So the process of joining the quorum would have to be incremental. Because demanding a unanimous vote to add a machine means you can never replace a dead one (except by impersonating it)
- benesch 7y ago
  > Should it not be the case that one of the responsibilities of the quorum is to vote new members in or out? I mean, as a first-class feature of the system. Indeed. What you're describing is one of the main motivations for Raft. Paxos showed that distributed consensus was mathematically sound, but did little to guide implementors in actually building such a system. Raft is not a fundamentally new consensus algorithm; just an incremental improvement that formalizes many of the improvements that you needed to make to Paxos anyway, like membership changes, log compaction, and multi-decree support from the get go. If you're interested, the Raft paper is quite readable and goes into this in detail. [0] > The main problem I see is that in most consensus systems, any member can be nominated as leader. But if the member was inducted during a partition event - which one could do if the partition were long duration - then nominations for new leaders will go out. What happens if the new member gets elected? How do the partitioned machines find that leader when they return? How do the partitioned machines find that leader when they return? This isn't actually the tricky bit, as it turns out. Communication flows from the leader to the other nodes, so the leader, even if it's a newly-inducted node, will initiate the connections to the partitioned nodes when the partition resolves. (The leader necessarily knows the addresses of the partitioned nodes, because the leader knows about all committed entries, and the identities of all the nodes in the cluster are committed into the Raft log.) Again, the Raft paper does a great job explaining cluster mebership changes—much better than I can! [0]: https://raft.github.io/raft.pdf https://raft.github.io/raft.pdf

WaxProlix 7y ago

Relieving to hear this as I've had almost exactly the same experience and it made me feel like an actual idiot.

necubi 7y ago

Using ASGs with zookeeper is kind of a pain, but also not very necessary. Production zookeeper clusters are almost always either 3, 5, or 7 nodes and it's very rare to change that once a cluster is in production. Given that this is all very static, the easiest thing is to just use terraform or similar to create the ec2 instances. If a node dies, you re-run your terraform to create a replacement. Alternatively, you can create an ASG-per-node so that you get auto-replacement. In my experience, ZK is one of the easiest and most reliable distributed systems to operate. I've only seen issues when it's used as a database instead of a distributed coordination service.

coredog64 7y ago
We ran into additional troubles based on ZK client bugs in the version that Kafka used. It would only ever resolve the ZK hostname once, at startup. Our workaround required some additional work to allocate an EIP for each host (3 AZs -> 3 EIPs) and then have the ZK host grab the EIP on startup. Even though Netflix hasn't updated it in a while, Exhibitor was helpful as well, in that it allowed ZK to bootstrap nodes off of state stored in S3. That did come at the cost of an extra 2-3 minutes per node on initial quorum startup.
- uberduper 7y ago
  Might just need to set networkaddress.cache.ttl=60 in java.security.

fizx 7y ago

ASGs + Zookeeper is really easy, but you have to use https://github.com/soabase/exhibitor https://github.com/soabase/exhibitor. It coordinates replacing downed ZK nodes using S3, and also provides a bunch of extra management tools. I had a single Zookeeper ASG running under load for over three years without maintenance. I pinged my old co-founder to see if he's willing to open source the CloudFormation template.

RantyDave 7y ago

Yes, it was. More specifically Zookeeper is designed as a piece of infrastructure you put together then rely on.