09:13 PM. Here those three paritions lag is increased without cleared at DEF topics and remaining parition is showing lag is zero. 09:02 AM, FLUME-3027 has been backported to CDH5.11.0 and above, so if you are able to upgrade, it would prevent the issue of offsets bouncing back and forward.One thing you may want to consider, if you are getting rebalances, it may be because it is taking too long to deliver by your sink, before polling kafka again. CDP Operational Database (COD) supports CDP Control Planes for multiple regions. Created on Then data is comming into hive tables. No error or exceptions apart from above messages. What was wrong in my configuration. ------------------------------------------, # Channelsagent.channels = \kafkachannel \, # Sinksagent.sinks = \kite-sink-1 kite-sink-2 \, # Sourcesagent.sources.flume-agent.channels = kafkachannelagent.sources.flume-agent.type = org.apache.flume.source.kafka.KafkaSourceagent.sources.flume-agent.kafka.bootstrap.servers = example1.host.com:9092,example2.host.com:9092,example3.host.com:9092agent.sources.flume-agent.kafka.num.consumer.fetchers = 10agent.sources.flume-agent.kafka.topics = abcagent.sources.flume-agent.interceptors = abc-interceptor abcinterceptoragent.sources.flume-agent.interceptors.abc-parameters-interceptor.type = staticagent.sources.flume-agent.interceptors.abcinterceptor.type = com.abc.flume.interceptor.ABCinterceptor$Builderagent.sources.flume-agent.interceptors.abc-parameters-interceptor.key = flume.avro.schema.urlagent.sources.flume-agent.interceptors.abc-parameters-interceptor.value = hdfs://myhadoop/abc.avscagent.sources.flume-agent.interceptors.abc-parameters-interceptor.threadNum = 10agent.sources.flume-agent.kafka.consumer.security.protocol = PLAINTEXTagent.sources.flume-agent.kafka.consumer.group.id = my-group, # Sinksagent.sinks.kite-sink-1.channel = kafkachannelagent.sinks.kite-sink-1.type = org.apache.flume.sink.kite.DatasetSinkagent.sinks.kite-sink-1.kite.repo.uri = repo:hiveagent.sinks.kite-sink-1.kite.dataset.name = abcagent.sinks.kite-sink-1.kite.batchSize = 100000agent.sinks.kite-sink-1.kite.rollInterval = 30agent.sinks.kite-sink-2.channel = kafkachannelagent.sinks.kite-sink-2.type = org.apache.flume.sink.kite.DatasetSinkagent.sinks.kite-sink-2.kite.repo.uri = repo:hiveagent.sinks.kite-sink-2.kite.dataset.name = abcagent.sinks.kite-sink-2.kite.batchSize = 100000agent.sinks.kite-sink-2.kite.rollInterval = 30, # Channelsagent.channels.kafkachannel.type = org.apache.flume.channel.kafka.KafkaChannelagent.channels.kafkachannel.brokerList = example1.host.com:9092,example2.host.com:9092,example3.host.com:9092agent.channels.kafkachannel.kafka.topic = defagent.channels.kafkachannel.kafka.consumer.group.id = my-group-kiteagent.channels.kafkachannel.parseAsFlumeEvent = trueagent.channels.kafkachannel.kafka.consumer.session.timeout.ms =60000agent.channels.kafkachannel.kafka.consumer.request.timeout.ms=70000agent.channels.kafkachannel.kafka.consumer.max.poll.records=100000agent.channels.kafkachannel.kafka.num.consumer.fetchers = 10. kafka server.properties modified than defaults: Kindly help me. 08-24-2018 08-24-2018 Auto-suggest helps you quickly narrow down your search results by suggesting possible matches as you type. 11:08 AM
Strangely lag data is flushed but remaining partition reset current-offset to previous offset even though no data in those particular partitions. Kafka consumer group lag in one or two partition e https://issues.apache.org/jira/browse/FLUME-3027, CDP Public Cloud Release Summary: June 2022, Cloudera DataFlow for the Public Cloud 2.1 introduces in-place upgrades as a Technical Preview feature. It did not cleared in next a day or two days also. Either we have to upgrade to 1.8 of flume or adding offsets.clear() code end of the method. Created 08-31-2018 - edited 08-31-2018 Created after one or two mins. Kafka ABC topic data is cleared and but not at DEF topics of 1,4 and 6 partitions even though data is not coming into system. we observed either one or two or three paritions (assume 1,4 ,6) at DEF topic is showing as LAG is increasing constantly compare with flume source.But strangely kafka ABC (three partition) is working fine and lag is showing equally. We restarted kafka and flume couple of times. Kafka consumer group lag in one or two partition even without data ingestion into kafka. 08-27-2018 Additionally, you could increase the max.poll.interval.ms, which is decoupled from the session.timeout.ms in 0.10.0 and above. 08-31-2018 We stopped all flume agent an hour or two hours and started data ingestion around 12lacs per min into kafka ABC topics. Meanwhile we have separate cluster where data lake is running ( HDFC,Hive, HBase, Cloudera manager,Phoenix client). I observed strange behaviour ingested data around 50 records into sytem when we have data lag at Kafka channel topic. I have a strange problem at kafka channel topic like kafka consumer group lag( 15 lacs events) in one or two partition only.I'll give little background aboout problem: Please find the data flow into system as shown below: data ingestion ==> kafka ABC(topic of 3 parition) ==> flume source (interceptor ) ==> Kafka DEF(topic of 6 partition ) ==> Hive (parquet file), This is three node edge nodes ( kafka, flume) are running in all nodes. we stopped data ingestions into system and started all flume agents (in three nodes). 04:25 AM. please see offset reset to earlier below screenshot. Find answers, ask questions, and share your expertise.
08:23 PM, I did not observe any error message in kafka or flume logs.But, i saw below messages in flume logs, 2018-08-24 04:35:47 DEBUG Fetcher:677 - Discarding stale fetch response for partition def-3 since its offset 395690 does not match the expected offset 3941182018-08-24 04:35:47 DEBUG Fetcher:677 - Discarding stale fetch response for partition def-0 since its offset 81880195 does not match the expected offset 818783802018-08-24 04:35:47 DEBUG Fetcher:671 - Ignoring fetched records for partition def-4 since it is no longer fetchable2018-08-24 04:35:47 DEBUG Fetcher:671 - Ignoring fetched records for partition def-5 since it is no longer fetchable2018-08-24 04:35:47 DEBUG Fetcher:671 - Ignoring fetched records for partition def-1 since it is no longer fetchable, 2018-08-24 04:35:47 DEBUG Fetcher:677 - Discarding stale fetch response for partition def-3 since its offset 395690 does not match the expected offset 3941182018-08-24 04:35:47 DEBUG Fetcher:677 - Discarding stale fetch response for partition def-0 since its offset 81880195 does not match the expected offset 818783802018-08-24 06:11:03 DEBUG Fetcher:677 - Discarding stale fetch response for partition def-4 since its offset 482480 does not match the expected offset 482450. When usually current offset is going to reset previous/earlier position even though without any restart of flume component (i.e.consumer).
But sometimes one or two paritions lag is going to cleared, but not all paritions. Created on - edited 08-24-2018 This is a bug in flume 1.7 version (https://issues.apache.org/jira/browse/FLUME-3027).This issue is resolved with adding code offsets.clear() in below method after consumer.commitSync method otherwise it will create problem when consumer rebalancing time. 04:21 AM
- edited How it is possible reset consumer current-offset when we are using as flume sinks. This would prevent the rebalancing from occurring since the client would still heartbeat without having to do a poll to pull more records before session.timeout.ms expires.-pd, Created on CDP Operational Database (COD) supports Multiple Availability Zones (Multi-AZ) on AWS. 11:02 AM. You may want to consider lowering your sink batch size in order to deliver and ack the messages in a timely fashion.Additionally, if you upgrade to CDH5.14 or higher, the flume kafka client is 0.10.2, and you would be able to set max.poll.records to match the batchSize you are using for the flume sink. 08-31-2018 09:01 AM Sometimes, total data is reached at hive tables.But,lag will be remain same; why flume sink consumed offset is not updating at group coordinator level?
how to resolve consumer lag in kafka
