Can you just monitor the consumer byte/message/fetch rate? Thanks,
Jun On Thu, Nov 20, 2014 at 5:31 PM, Bhavesh Mistry <mistry.p.bhav...@gmail.com> wrote: > HI Jun, > > Do you want me to request Jira ticket for feature a notification for new > consumer API and old consumer feature that consumer stream is dying. So > application can try to restart it programmatically. I understand this is > due to network or zk cluster instability. > > Let me know if you have alternative proposal for this for new and old > high-level consumer API. > > Thanks, > > Bhavesh > > On Tue, Nov 18, 2014 at 9:53 PM, Bhavesh Mistry < > mistry.p.bhav...@gmail.com> > wrote: > > > Hi Jun, > > > > ZK cluster are up and running. What is best way to programmatically > > recover and I would try to exponential recovery process which I am > willing > > to implement. So do you think monitoring "ZkClient-EventThread > > <http://zkclient-eventthread-65-dare-msgq00.sv.walmartlabs.com:9091/>*" > > thread status will be enough to indicate source thread is dead and > > therefore start exponential reconnect process ? > > > > Can you guys at least for new consumer api (0.9.0) provide a call back > > method or notification the consumer is died and reason for it ? > > > > > > Thanks, > > Bhavesh > > > > > > > > On Tue, Nov 18, 2014 at 9:34 PM, Jun Rao <jun...@gmail.com> wrote: > > > >> Is your ZK service alive at that point? If not, you just need to monitor > >> the ZK server properly. > >> > >> Thanks, > >> > >> Jun > >> > >> On Mon, Nov 17, 2014 at 2:30 PM, Bhavesh Mistry < > >> mistry.p.bhav...@gmail.com> > >> wrote: > >> > >> > Hi Kafka Team, > >> > > >> > > >> > I get following exception due to ZK/Network issues intermittently. > How > >> do > >> > I recover from consumer thread dying *programmatically* and restart > >> source > >> > because we have alerts that due to this error we have partition > >> OWNERSHIP > >> > is *none* ? Please let me know how to restart source and detect > >> consumer > >> > thread died and need to be restarted ? > >> > > >> > > >> > > >> > 17 Nov 2014 04:29:41,180 ERROR [ > >> > ZkClient-EventThread-65-dare-msgq00.sv.walmartlabs.com:9091, > >> > dare-msgq01.sv.walmartlabs.com:9091, > dare-msgq02.sv.walmartlabs.com:9091 > >> ] > >> > (org.I0Itec.zkclient.ZkEventThread.run:77) - Error handling event > >> > ZkEvent[New session event sent to > >> > > >> > kafka.consumer.ZookeeperConsumerConnector$ZKSessionExpireListener@552c5ea8 > >> > ] > >> > kafka.common.ConsumerRebalanceFailedException: > >> > > >> > > >> > mupd_statsd_logmon_metric_events18_sdc-stg-logsearch-app1-1416023311208-14501d68 > >> > can't rebalance after 8 retries > >> > at > >> > > >> > > >> > kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:626) > >> > at > >> > > >> > > >> > kafka.consumer.ZookeeperConsumerConnector$ZKSessionExpireListener.handleNewSession(ZookeeperConsumerConnector.scala:481) > >> > at org.I0Itec.zkclient.ZkClient$4.run(ZkClient.java:472) > >> > at > org.I0Itec.zkclient.ZkEventThread.run(ZkEventThread.java:71) > >> > > >> > > >> > > >> > > >> > > >> > ZK Connection Issues: > >> > > >> > java.net.SocketException: Transport endpoint is not connected > >> > at sun.nio.ch.SocketChannelImpl.shutdown(Native Method) > >> > at > >> > sun.nio.ch.SocketChannelImpl.shutdownInput(SocketChannelImpl.java:633) > >> > at > >> sun.nio.ch.SocketAdaptor.shutdownInput(SocketAdaptor.java:360) > >> > at > >> > > org.apache.zookeeper.ClientCnxn$SendThread.cleanup(ClientCnxn.java:1205) > >> > at > >> > org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1170) > >> > > >> > > >> > > >> > > >> > at > >> > org.I0Itec.zkclient.exception.ZkException.create(ZkException.java:47) > >> > at > >> > org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:685) > >> > at org.I0Itec.zkclient.ZkClient.readData(ZkClient.java:766) > >> > at org.I0Itec.zkclient.ZkClient.readData(ZkClient.java:761) > >> > at kafka.utils.ZkUtils$.readData(ZkUtils.scala:449) > >> > at > >> > kafka.consumer.TopicCount$.constructTopicCount(TopicCount.scala:61) > >> > at > >> > > >> > > >> > kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.kafka$consumer$ZookeeperConsumerConnector$ZKRebalancerListener$$rebalance(ZookeeperConsumerConnector.scala:630) > >> > at > >> > > >> > > >> > kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1$$anonfun$apply$mcV$sp$1.apply$mcVI$sp(ZookeeperConsumerConnector.scala:601) > >> > at > >> scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:141) > >> > at > >> > > >> > > >> > kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply$mcV$sp(ZookeeperConsumerConnector.scala:595) > >> > at > >> > > >> > > >> > kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply(ZookeeperConsumerConnector.scala:592) > >> > at > >> > > >> > > >> > kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply(ZookeeperConsumerConnector.scala:592) > >> > at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33) > >> > at > >> > > >> > > >> > kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:591) > >> > at > >> > > >> > > >> > kafka.consumer.ZookeeperConsumerConnector$ZKSessionExpireListener.handleNewSession(ZookeeperConsumerConnector.scala:481) > >> > at org.I0Itec.zkclient.ZkClient$4.run(ZkClient.java:472) > >> > at > org.I0Itec.zkclient.ZkEventThread.run(ZkEventThread.java:71) > >> > Caused by: org.apache.zookeeper.KeeperException$NoNodeException: > >> > KeeperErrorCode = NoNode for > >> > > >> > > >> > /consumers/mupd_statsd_logmon_metric_events18/ids/mupd_statsd_logmon_metric_events18_sdc-stg-logsearch-app1-1416023311208-14501d68 > >> > at > >> > org.apache.zookeeper.KeeperException.create(KeeperException.java:102) > >> > at > >> > org.apache.zookeeper.KeeperException.create(KeeperException.java:42) > >> > at org.apache.zookeeper.ZooKeeper.getData(ZooKeeper.java:921) > >> > at org.apache.zookeeper.ZooKeeper.getData(ZooKeeper.java:950) > >> > at > >> org.I0Itec.zkclient.ZkConnection.readData(ZkConnection.java:103) > >> > at org.I0Itec.zkclient.ZkClient$9.call(ZkClient.java:770) > >> > at org.I0Itec.zkclient.ZkClient$9.call(ZkClient.java:766) > >> > at > >> > org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:675) > >> > > >> > > > > >
