BDSBTB 2015: Hiral Patel, Managing Kafka, the easy way, with help from Play, Akka, and Curator
okay so we're going to be talking about managing kaka the easier the easy way with the help of play acha and curator let me introduce myself I'm a principal engineer at Yahoo I've worked on real-time analytics pipelines and currently working on reporting for Jim and I our search and native ads platform work that big data companies in the bay using both open open source as well as enterprise solutions and I've worked with Scott various companies in the bay now how many of you are already using kafka in production all right that's like sixty seventy percent of you and how many of you are looking to use calf grand production okay the rest of you okay so let's get started so I'm going to spend very little time look doing an overview of calf guys so everyone knows what kalka is some of the operational challenges from my perspective and why should you use calcio manager and how it works internally so let's take an example cluster here we have a three node cluster and we want to create a topic called test with three partitions with a replication factor of two so you can see we have p0 p1 and p2 those are the three partitions here you can see what the JSON representation might be for the partition assignments which you can see for partition 0 we have brokers 1 & 2 right here we have r1 r2 for p0 where we're Burger one is the leader the leader is responsible for all reads and writes to that partition and be the other replicas pull the leader for updates and you you see that you have the producers on this side and the consumers on this on the site and the consumers are going to be talking to always going to be talking to the leaders and the producers are always going to be writing to the leaders so this is a high-level overview of how topic is looks like on a capital cluster so some of the operational challenges around Kafka are related to skew one form of ski was dat askew so the data being written to the cluster to the topics it may not be evenly distributed across all your partition so this can happen if you're using partitioning by key or value and this manifests in in in the network is a network I owe on on some of your host will be really high and your other house will be very low right and so there's no easy way to spot this without looking at looking at the actual sizes of each partition another source of data skew could be that some of the some of the messages that you're writing to the partitions are really small while others are really large and so there's no way for for you to really distribute the load evenly unless you look at the size of the partitions and do some sort of batching on your own another operational challenge around Kafka is leader skew so this is something that isn't a problem if you already have automatic preferred leader election enabled in Kafka this is this goes back to these leaders that we have here one two and three it's entirely possible for for some out for some reason there's an outage and the third broker the burqa three becomes a leader here but it's not the preferred leader so in Kafka you you have to run the preferred leader election or have that enable where it can do it automatically right so when you don't have it enabled automatically this was what what happens to your network I oh alright so you have a bunch of brokers where which have leaders on them and you have a bunch of brokers that have very very few leaders on them so your network i/o is not even across your cluster and as soon as you run preferred leader election all the network by realizes now the source of ski was is replica skew so let's say you have a ten node cluster and you have n number of topics that are distributed across those Cluttons those notes and you add of five more notes and I have 15 nodes but the fifth the five new nodes have no data on them right now right because partitions need to be reassigned such that they used on you new set of brokers so this is called part this is called the reassignment Partridge partition or assignment and first you need to generate the set of partitions with the 15 brokers before you reassign them so again this is this the same kind of network I owe that you would see when you don't have replicas as evenly distributed another reason you may not have replicas evenly distributed as if you have 10 brokers and you're the number of partitions you've created for the topic is like to write so how are you going to evenly divided two partitions across 10 10 10 notes right so that's another source of replicas q so as soon as you run a reassignment which which is what the spike is where it's it's copying all the data across all the brokers the network I go io after the operation is done normalizes across across your cluster so think of Kafka manager as a mission control for Kafka except for less complicated right out of the box you get all the functionality you would get from the kafka command line utilities except where you have a nice UI to do this with right so let's quickly look at the UI so I can put some context around what these things are all right so let's see okay so this is the the broker list view this lists all the brokers that are that are available and all the metrics associated with these brokers so this is one of our production clusters as you can see here you have bites in and bites out it's doing in the last minute hundred two hundred sixty megabytes in and two gigabytes per second out alright so that's about eighteen gigabits that's going through this cluster right now we look at the topic view here and the topic view essentially gives you the information about about the topic it also tells you if there are any topic any many partitions which don't have the preferred replicas as the leader and so as you can see there's there about twelve percent of the the partitions don't have the preferred replicas as as the leader and it tells you if there's any burgers queue or burger the burger spread percent essentially tells you how many what percentage of your brokers is the topic spread across right so these are just basic metrics that you would need to optimize the way that you have the topics laid out on your cluster some of the operations you can do delete topic are at partitions you can update the configuration you can earn the recent on the on the topic you can generate partition assignments so if I were to click on this you you would see this and this is basically a a confirmation page where you can select which topic or which set of brokers you want to put this topic on so let's say if you had an operation where you have 10 hosts and you want to add a new host and remove one of the old hosts right so you would basically go to this page unselect the one you want to remove add the one that you want to want to move the part in the new data to add the new one that you have added in the list and regenerate the partitions and then go back to this page here and and run the run the partition assignment and this is what the update configuration page looks like so you don't have to go look up all the different config options for Kafka they're all just listed out here and you can just fill in whatever you need an update update the configuration for that topic so let's go back to the presentation a few other things that you get with the CAF commander jar when two concurrent updates are happening if two users are trying to modify the same exact topic config at the same time the command line utility would just let that happen right so the second user would not see what the change the first user had applied and the second change we just override the the first change right so it with calcio manager we can gate the updates because we can keep a track of the zenoid version that we had when we read the topic the Indian the initial topic config right so this this makes it a little safer to update topic configuration if you know that two people cannot overwrite each other's changes right another thing that you get with capital managers the ability to block generate assignments command for topics that are currently undergoing or assignment so what happens or what could happen is if you have a topic let's say that has a replication of three and you generated new partition assignments for it and ran the resigned partitions command while it's undergoing the reassignment the number of replicas that are assigned to a partition could jump from three to five because it may be moving data between brokers right so temporarily its replication factor as far as cough condos it is five because there's no place where it stores the actual value of the replication factor it's entirely determined based on the number of replicas that are assigned to a partition right so if you generated assignments at that time all of a sudden your assignments would would generate with a replication factor of five instead of three like you want it right so we can get this because in calcio manager we know what topics are currently going undergoing reassignment socal co-manager internally is a play framework application so all the UI is based on play the play framework is all the controls are written in play of course and the controls are interacting with a class called Capcom manager which provides an API to the actor hierarchy so there's a clear separation between the UI and actual functionality so at any point we can replace the UI with any other UI or put it on under arrest api if we wanted to all the interactions with zookeeper are done using patch curator the way we manage state in cal commanders distort and zookeeper we didn't want to introduce yet another data store to keep configuration information about the clusters that you want to add to calculator so every time you add a new cluster we actually store that information in zookeeper and every time you generate a partition assignment we save that until inside a zookeeper for Kafka itself all the state is in zookeeper and what we do is we use curators treeck treeck ash and the path children cash patterns to actually cash the zookeeper state in memory right so we don't have to make calls to zookeeper every time we want to see the number of brokers that are currently in the cluster right we query this in-memory cache and this is a very great pattern that is available through curator another reason why we're using it so the actor model that we've defined has a logical separation between command and query processing so this is inspired by seek urs if you were at guys are here for a scala by the bay you may have heard some sakura stalks and this is essentially the pattern that we're using here we have separate processing for commands and queries so when we want to make extensions to cap the manager it's easy to do because we know what happened what we need to do when we want to add a command functionality or a query functionality and command query basically means the right side is the command side and the query side is the is the read side of of the functionality this allows us to to build some interesting abstractions on top to make things easier to to extend Kafka manager itself so we can create a base actor which is responsible for processing the actor request an actor response and then we can create a base command actor which extends the base actor and all its doing is processing a command request and then we can do the same thing with the query side and create a base query command actor which process the query requests as well as a command request and then we can use these abstractions to build up our actor hierarchy the apt so we start with a calf co-manager actor which is responsible for instantiating all the other actors so capital manager actor is what connects the zookeeper gets the cluster information and instantiates all the cluster manager actors the cluster manager so we have a cluster manager actor for every cluster that you've defined in calculator the delete cluster actor is responsible for cleaning up state and zookeeper so when you go in there and remove a cluster we have all the generated generator partition assignment information stored in zookeeper so when you delete a cluster this delete cluster actor goes in there and cleans up that state that that's left over when you delete the cluster the cluster manager actor itself instantiates three other actors the broker view cash actor the CAF got state actor in the CAF kokum and actor so all changes to zookeeper are go through the CAF command actor which is the right side of the SI quieres pattern and all reads for state go to calf cast a doctor and the broker view cash actor essentially is all the views that you see in the UI are maintained in this actor so this actor periodically queries the calc estate actor for the current state and builds the necessary views it's also responsible for doing all the gym xquery to all the brokers and merging that data into its views so when we look at the metrics that's that's actor that's responding with the data so some of the work that is currently going on is related to smart reassignment so when you generate partition assignments right now it's entirely possible that you move data from one set of brokers to another set of holes because when the command itself randomly picks a broker to start from right so it doesn't take into consideration that there are there is a set of brokers that already have this data and they should not move it should not be moved to another set of brokers right and this is where we want to reduce data movement and not have to worry about too much network i/o happening especially if you're doing hi months of network i/o like 18 gigabits right you don't want to move data around when you don't need to partition log segment sizes is a working progress as well that will allow us to see the actual sizes of the data on disk for each partition so this is where you'll be able to see what the data skew actually is a cross your partition sore topic and consumer offsets are a feature that I've been requested by a lot of people and there are two PRS for that oh that are already there for the seizure and we're just going through figuring out which one we're going to merge and just going through the code review process right now there are a few other features that we'd like to have and we'd love to have more contributions to calculor it's like it's open sourced it's on our github page and we would be more than happy to have more contributions so some of the contributors that we've already had our from meta markets ji sook Kim John and Alexander have worked with us collaboratively to add new features to kafka manager and a few other people so contributions are welcomed if you have any questions we know alright so the question was what is meant by the consumer offsets work so we basically are going to add a UI page where you can see the list of consumers and were where they are in consuming data from the topic so you can see what office upset they're currently at this is basically there to identify lagging consumers you