Kafka-code-Notes

Kafka-code-Notes

Project Dirs


 bin : Scripts to manage kafka and Zookeeper
 checkstyle : Checkstyle configurations
 clients : Kafka client code [Producer,Consumer, Admin tools classes]
 config : Default Config files
 connect : Kafka Connect module
 core : Kafka server code
 docs : Documentations
 examples : Producr , Consumer examples
 gradle : Build support
 JMH-Benchmark: benchmarks written using [JMH](http://openjdk.java.net/projects/code-tools/jmh/)
 log4j-appender : Kafka log4j Extension
 streams : Kafak Stream module
 tests : Kafka test tools
 tools : ???
 vagrant : Kafak with Vagrant
 

 
 Commands
 -----------------
 
 1) Send key, value from kafka console producer
 ./kafka-console-producer.sh --broker-list localhost:9092 --topic t1 --property "parse.key=true" --property "key.separator=:"

2) View key value from kafka console consumer
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic t1 --property print.key=true  --property key.separator="-" --from-beginning



Code flow
------------------

Kafka->KafkaServerStartable.startup -> KafkaServer.startup -----
                                                           setup zookeeper
                                                           Get or create cluster_id
                                                           generate brokerId
                                                           initialize dynamic broker configs from ZooKeeper
                                                           start scheduler
                                                           create and configure metrics
                                                           register broker metrics
                                                           start log manager
                                                           Create and start the socket server acceptor threads 
                                                           start replica manager
                                                           Register registerBroker to zk
                                                           start token manager
                                                           start kafka controller
                                                           start group coordinator
                                                           start transaction coordinator
                                                           start processing requests
                                                           start dynamic config manager
                                                           Create the config manager. start listening to notifications
                                                           socketServer.startDataPlaneProcessors()
                                                           socketServer.startControlPlaneProcessor()
                                                           brokerState.newState(RunningAsBroker)
                                                           shutdownLatch = new CountDownLatch(1)
                                                           startupComplete.set(true)
                                                           isStartingUp.set(false)