Also, I ran the flume in debug mode : [cloudera@quickstart flume-ng]$ ./bin/flume-ng agent --conf conf -conf-file conf/flumekafka.conf --name agent1 -Dflume.root.logger=DEBUG,console   Getting the below snippet repeatedaly with 'Updated cluster metadata version' value changing 2018-03-16 12:01:51,797 (lifecycleSupervisor-1-4) [DEBUG - org.apache.kafka.clients.consumer.internals.AbstractCoordinator.sendGroupCoordinatorRequest(AbstractCoordinator.java:470)] Sending coordinator request for group flume to broker quickstart.cloudera:9092 (id: 33) 2018-03-16 12:01:51,801 (lifecycleSupervisor-1-4) [DEBUG - org.apache.kafka.clients.consumer.internals.AbstractCoordinator.handleGroupMetadataResponse(AbstractCoordinator.java:483)] Received group coordinator response ClientResponse(receivedTimeMs=1521226911801, disconnected=false, request=ClientRequest(expectResponse=true, callback=org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler@10f78e4b, request=RequestSend(header={api_key=10,api_version=0,correlation_id=15,client_id=consumer-1}, body={group_id=flume}), createdTimeMs=1521226911798, sendTimeMs=1521226911798), responseBody={error_code=15,coordinator={node_id=-1,host=,port=-1}}) 2018-03-16 12:01:51,879 (lifecycleSupervisor-1-4) [DEBUG - org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeUpdate(NetworkClient.java:627)] Sending metadata request ClientRequest(expectResponse=true, callback=null, request=RequestSend(header={api_key=3,api_version=0,correlation_id=16,client_id=consumer-1}, body={topics=[Airports]}), isInitiatedByNetworkClient, createdTimeMs=1521226911878, sendTimeMs=0) to node 33 2018-03-16 12:01:51,891 (lifecycleSupervisor-1-4) [DEBUG - org.apache.kafka.clients.Metadata.update(Metadata.java:180)] Updated cluster metadata version 10 to Cluster(nodes = [quickstart.cloudera:9092 (id: 33)], partitions = [Partition(topic = Airports, partition = 6, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 10, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 1, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 11, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 13, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 5, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 7, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 12, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 8, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 17, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 16, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 4, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 3, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 14, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 2, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 15, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 19, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 18, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 0, leader = 33, replicas = [33,], isr = [33,], Partition(topic = Airports, partition = 9, leader = 33, replicas = [33,], isr = [33,]]) S ... View more

Also some character case issue, you should follow flume official document to do configure. http://flume.apache.org/FlumeUserGuide.html#spooling-directory-source ... View more