Syslog -> Flume Agent -> Kafka -> Kafka Mirror Maker

gistfile1.txt

# Configure Flume agent (/etc/flume/conf/flume.conf) to receive syslog messages 
agent.sources=syslogsource-1
agent.channels=mem-channel-1
agent.sinks=kafka-sink-1

agent.sources.syslogsource-1.type=syslogtcp
agent.sources.syslogsource-1.port=13073
agent.sources.syslogsource-1.host=0.0.0.0
agent.sources.syslogsource-1.channels=mem-channel-1

agent.channels.mem-channel-1.type = memory

agent.sinks.kafka-sink-1.channel = mem-channel-1
agent.sinks.kafka-sink-1.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.kafka-sink-1.batchSize = 5
agent.sinks.kafka-sink-1.brokerList = kafka_host:6667
agent.sinks.kafka-sink-1.topic = test.topic

# start flume agent 
flume-ng agent --conf-file  /etc/flume/conf/flume.conf --name agent 

# create kafka topic
/usr/hdp/current/kafka-broker/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor=1 --partitions 1 --topic test.topic

# Prepare to watch data via kafka cosole consumer
/usr/hdp/current/kafka-broker/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test.topic --from-beginning

# Generate some syslog messages 
yum -y install syslog-ng
loggen -I 113 -s 113 -r 10 flume_host 13073

# Mirror from one Kafka cluster to another
# consumer.properties
zookeeper.connect=zk_cluster1:2181
zookeeper.connection.timeout.ms=1000000
group.id=test-consumer-group

# producer.properties
metadata.broker.list=kafka_broker2:6667
zk.connect=zk_cluster2:2181
producer.type=sync
compression.codec=none
serializer.class=kafka.serializer.DefaultEncoder

/usr/hdp/current/kafka-broker/bin/kafka-run-class.sh kafka.tools.MirrorMaker --consumer.config consumer.properties --num.streams 1 --producer.config producer.properties --whitelist test.topic