[FLUME-2870] Flume agent throw NULLPOINTEXCEPTION when use kakfa-source

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Duplicate
• Affects Version/s: 1.6.0
• Component/s: Build
• Labels: kafka-source

Environment

kafka source,es sink

• Assignee: Denes Arvay
• Reporter: justdebugit
• Created: 28/Jan/16 04:17
• Updated: 28/Oct/22 13:29
• Resolved: 28/Oct/22 13:29

Description

CONFIG:
a1.sources = s1
a1.channels = c1
a1.sinks = k1

a1.sources.s1.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.s1.channels = c1
a1.sources.s1.zookeeperConnect = 10.xxx.xxx.160:5181,10.2x.xxx.203:5181/xxx/charge
a1.sources.s1.topic = charge.log
a1.sources.s1.groupId = flume

#a1.channels.c1.type = memory
#a1.channels.c1.capacity = 50000
#a1.channels.c1.transactionCapacity = 100

a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /home/xxxx/app/flume/checkpoint
a1.channels.c1.dataDirs = /home/xxxxx/app/flume/data
a1.channels.c1.capacity = 5000000

a1.sinks.k1.type = com.frontier45.flume.sink.elasticsearch2.ElasticSearchSink
a1.sinks.k1.hostNames = 127.0.0.1:9300
a1.sinks.k1.indexName = xxx_charge_log
a1.sinks.k1.indexType = logs
a1.sinks.k1.clusterName = elasticsearch
a1.sinks.k1.batchSize = 500
a1.sinks.k1.ttl = 3d
a1.sinks.k1.serializer = com.frontier45.flume.sink.elasticsearch2.ElasticSearchLogStashEventSerializer
a1.sinks.k1.channel = c1

EXCEPTION:

28 Jan 2016 11:09:03,830 ERROR [PollableSourceRunner-KafkaSource-s1] (org.apache.flume.source.kafka.KafkaSource.process:153) - KafkaSource EXCEPTION, {}
java.lang.NullPointerException
at org.apache.flume.instrumentation.MonitoredCounterGroup.increment(MonitoredCounterGroup.java:261)
at org.apache.flume.instrumentation.kafka.KafkaSourceCounter.incrementKafkaEmptyCount(KafkaSourceCounter.java:49)
at org.apache.flume.source.kafka.KafkaSource.process(KafkaSource.java:146)
at org.apache.flume.source.PollableSourceRunner$PollingRunner.run(PollableSourceRunner.java:139)
at java.lang.Thread.run(Thread.java:745)