It is accurate since it’s an API/implementation problem and therefore container independent. Sure if everything is configured correctly and broker is accessible then things do work, but try to shut down consumer when broker is not accessible. And when I mean shut down I am not implying shutting down the JVM via System.exit(), I am simply saying that poll() will block indefinitely and so will close().
In fact it’s a very common problem in Kafka components (see below) https://issues.apache.org/jira/browse/KAFKA-3540 https://issues.apache.org/jira/browse/KAFKA-1894 https://issues.apache.org/jira/browse/KAFKA-3539 Cheers Oleg On Aug 16, 2016, at 5:06 AM, Jaikiran Pai <jai.forums2...@gmail.com<mailto:jai.forums2...@gmail.com>> wrote: On Friday 12 August 2016 08:45 PM, Oleg Zhurakousky wrote: It hangs indefinitely in any container. I don't think that's accurate. We have been running Kafka brokers and consumers/producers in docker containers for a while now and they are functional. Of course, you need to make sure you use the IP addresses instead of localhost/127.0.0.1 to make sure that the brokers are accessible to the consumers/producers and you don't run into the situation that you explain about the broker connection not happening successfully. By the way, I am not saying that the consumer.poll() doesn't have that issue you state. -Jaikiran It’s a known issue and has been brought up many times on this list, yet there is not fix for it. The problem is with the fact that while poll() attempts to create an elusion that it is async and even allows you to set a timeout it is essentially very misleading if you look inside its implementation. The first call it makes is to fetch topic metadata. That call is not part of the Future it returns so if connection to broker is not available you’re dead since Kafka attempts to reconnect and there s no property to set reconnect attempts, so it attempts to reconnect indefinitely. Cheers Oleg On Aug 12, 2016, at 9:22 AM, Brem, Robert <robert.b...@adesso.ch<mailto:robert.b...@adesso.ch>> wrote: Hy I'm new to Kafka and messaging at all. I have a simple java application that contains a consumer and a producer. It is working on the host system but if I try to run it in a docker container (Kafka is not in the container, it is still on the host) consumer.poll() hangs up and does not return. telnet tells me that inside the container the host:port 172.17.0.1:9092 is open. In the docker container on startup Kafka tells me: Marking the coordinator ... dead for group ... Can you give me a hint, in which direction I should look? Thanks! That's the output on the host, with the working application: 15:03:42,657 INFO [org.apache.kafka.clients.consumer.ConsumerConfig] (ServerService Thread Pool -- 63) ConsumerConfig values: metric.reporters = [] metadata.max.age.ms = 300000 partition.assignment.strategy = [org.apache.kafka.clients.consumer.RangeAssignor] reconnect.backoff.ms = 50 sasl.kerberos.ticket.renew.window.factor = 0.8 max.partition.fetch.bytes = 1048576 bootstrap.servers = [172.17.0.1:9092] ssl.keystore.type = JKS enable.auto.commit = true sasl.mechanism = GSSAPI interceptor.classes = null exclude.internal.topics = true ssl.truststore.password = null client.id = consumer-1 ssl.endpoint.identification.algorithm = null max.poll.records = 2147483647 check.crcs = true request.timeout.ms = 40000 heartbeat.interval.ms = 3000 auto.commit.interval.ms = 5000 receive.buffer.bytes = 65536 ssl.truststore.type = JKS ssl.truststore.location = null ssl.keystore.password = null fetch.min.bytes = 1 send.buffer.bytes = 131072 value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer group.id = pokertracker retry.backoff.ms = 100 sasl.kerberos.kinit.cmd = /usr/bin/kinit sasl.kerberos.service.name = null sasl.kerberos.ticket.renew.jitter = 0.05 ssl.trustmanager.algorithm = PKIX ssl.key.password = null fetch.max.wait.ms = 500 sasl.kerberos.min.time.before.relogin = 60000 connections.max.idle.ms = 540000 session.timeout.ms = 10000 metrics.num.samples = 2 key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer ssl.protocol = TLS ssl.provider = null ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1] ssl.keystore.location = null ssl.cipher.suites = null security.protocol = PLAINTEXT ssl.keymanager.algorithm = SunX509 metrics.sample.window.ms = 30000 auto.offset.reset = latest 15:03:42,680 INFO [org.apache.kafka.common.utils.AppInfoParser] (ServerService Thread Pool -- 63) Kafka version : 0.10.0.0 15:03:42,680 INFO [org.apache.kafka.common.utils.AppInfoParser] (ServerService Thread Pool -- 63) Kafka commitId : b8642491e78c5a13 15:03:42,681 WARN [org.apache.kafka.common.utils.AppInfoParser] (ServerService Thread Pool -- 63) Error registering AppInfo mbean: javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=consumer-1 at com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:437) at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1898) at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:966) at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:900) at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:324) at com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522) at org.jboss.as.jmx.PluggableMBeanServerImpl$TcclMBeanServer.registerMBean(PluggableMBeanServerImpl.java:1527) at org.jboss.as.jmx.PluggableMBeanServerImpl.registerMBean(PluggableMBeanServerImpl.java:871) at org.apache.kafka.common.utils.AppInfoParser.registerAppInfo(AppInfoParser.java:58) at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:694) at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:587) at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:569) at com.optimist.pokerstats.pokertracker.kafka.control.KafkaProvider.createConsumer(KafkaProvider.java:44) at com.optimist.pokerstats.pokertracker.kafka.control.KafkaProvider.init(KafkaProvider.java:29) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at org.jboss.weld.injection.producer.DefaultLifecycleCallbackInvoker.invokeMethods(DefaultLifecycleCallbackInvoker.java:98) at org.jboss.weld.injection.producer.DefaultLifecycleCallbackInvoker.postConstruct(DefaultLifecycleCallbackInvoker.java:81) at org.jboss.weld.injection.producer.BasicInjectionTarget.postConstruct(BasicInjectionTarget.java:126) at org.jboss.weld.bean.ManagedBean.create(ManagedBean.java:171) at org.jboss.weld.context.AbstractContext.get(AbstractContext.java:96) at org.jboss.weld.bean.ContextualInstanceStrategy$DefaultContextualInstanceStrategy.get(ContextualInstanceStrategy.java:101) at org.jboss.weld.bean.ContextualInstanceStrategy$ApplicationScopedContextualInstanceStrategy.get(ContextualInstanceStrategy.java:141) at org.jboss.weld.bean.ContextualInstance.get(ContextualInstance.java:50) at org.jboss.weld.manager.BeanManagerImpl.getReference(BeanManagerImpl.java:742) at org.jboss.weld.injection.producer.AbstractMemberProducer.getReceiver(AbstractMemberProducer.java:123) at org.jboss.weld.injection.producer.AbstractMemberProducer.produce(AbstractMemberProducer.java:158) at org.jboss.weld.bean.AbstractProducerBean.create(AbstractProducerBean.java:181) at org.jboss.weld.context.unbound.DependentContextImpl.get(DependentContextImpl.java:70) at org.jboss.weld.bean.ContextualInstanceStrategy$DefaultContextualInstanceStrategy.get(ContextualInstanceStrategy.java:101) at org.jboss.weld.bean.ContextualInstance.get(ContextualInstance.java:50) at org.jboss.weld.manager.BeanManagerImpl.getReference(BeanManagerImpl.java:742) at org.jboss.weld.manager.BeanManagerImpl.getInjectableReference(BeanManagerImpl.java:842) at org.jboss.weld.injection.FieldInjectionPoint.inject(FieldInjectionPoint.java:92) at org.jboss.weld.util.Beans.injectBoundFields(Beans.java:378) at org.jboss.weld.util.Beans.injectFieldsAndInitializers(Beans.java:389) at org.jboss.weld.injection.producer.ResourceInjector$1.proceed(ResourceInjector.java:70) at org.jboss.weld.injection.InjectionContextImpl.run(InjectionContextImpl.java:48) at org.jboss.weld.injection.producer.ResourceInjector.inject(ResourceInjector.java:72) at org.jboss.weld.injection.producer.BasicInjectionTarget.inject(BasicInjectionTarget.java:121) at org.jboss.weld.bean.ManagedBean.create(ManagedBean.java:159) at org.jboss.weld.context.unbound.DependentContextImpl.get(DependentContextImpl.java:70) at org.jboss.weld.bean.ContextualInstanceStrategy$DefaultContextualInstanceStrategy.get(ContextualInstanceStrategy.java:101) at org.jboss.weld.bean.ContextualInstance.get(ContextualInstance.java:50) at org.jboss.weld.manager.BeanManagerImpl.getReference(BeanManagerImpl.java:742) at org.jboss.weld.manager.BeanManagerImpl.getInjectableReference(BeanManagerImpl.java:842) at org.jboss.weld.injection.FieldInjectionPoint.inject(FieldInjectionPoint.java:92) at org.jboss.weld.util.Beans.injectBoundFields(Beans.java:378) at org.jboss.weld.util.Beans.injectFieldsAndInitializers(Beans.java:389) at org.jboss.weld.injection.producer.DefaultInjector$1.proceed(DefaultInjector.java:71) at org.jboss.weld.injection.InjectionContextImpl.run(InjectionContextImpl.java:48) at org.jboss.weld.injection.producer.DefaultInjector.inject(DefaultInjector.java:73) at org.jboss.weld.injection.producer.ejb.SessionBeanInjectionTarget.inject(SessionBeanInjectionTarget.java:140) at org.jboss.as.weld.injection.WeldInjectionContext.inject(WeldInjectionContext.java:39) at org.jboss.as.weld.injection.WeldInjectionInterceptor.processInvocation(WeldInjectionInterceptor.java:51) at org.jboss.invocation.InterceptorContext.proceed(InterceptorContext.java:340) at org.jboss.as.ee.component.AroundConstructInterceptorFactory$1.processInvocation(AroundConstructInterceptorFactory.java:28) at org.jboss.invocation.InterceptorContext.proceed(InterceptorContext.java:340) at org.jboss.as.weld.injection.WeldInterceptorInjectionInterceptor.processInvocation(WeldInterceptorInjectionInterceptor.java:56) at org.jboss.invocation.InterceptorContext.proceed(InterceptorContext.java:340) at org.jboss.as.weld.ejb.Jsr299BindingsCreateInterceptor.processInvocation(Jsr299BindingsCreateInterceptor.java:100) at org.jboss.invocation.InterceptorContext.proceed(InterceptorContext.java:340) at org.jboss.as.ee.component.NamespaceContextInterceptor.processInvocation(NamespaceContextInterceptor.java:50) at org.jboss.invocation.InterceptorContext.proceed(InterceptorContext.java:340) at org.jboss.as.ejb3.tx.CMTTxInterceptor.invokeInOurTx(CMTTxInterceptor.java:275) at org.jboss.as.ejb3.tx.CMTTxInterceptor.requiresNew(CMTTxInterceptor.java:349) at org.jboss.as.ejb3.tx.LifecycleCMTTxInterceptor.processInvocation(LifecycleCMTTxInterceptor.java:66) at org.jboss.invocation.InterceptorContext.proceed(InterceptorContext.java:340) at org.jboss.as.weld.injection.WeldInjectionContextInterceptor.processInvocation(WeldInjectionContextInterceptor.java:43) at org.jboss.invocation.InterceptorContext.proceed(InterceptorContext.java:340) at org.jboss.as.ejb3.component.interceptors.CurrentInvocationContextInterceptor.processInvocation(CurrentInvocationContextInterceptor.java:41) at org.jboss.invocation.InterceptorContext.proceed(InterceptorContext.java:340) at org.jboss.as.ee.concurrent.ConcurrentContextInterceptor.processInvocation(ConcurrentContextInterceptor.java:45) at org.jboss.invocation.InterceptorContext.proceed(InterceptorContext.java:340) at org.jboss.invocation.ContextClassLoaderInterceptor.processInvocation(ContextClassLoaderInterceptor.java:64) at org.jboss.invocation.InterceptorContext.proceed(InterceptorContext.java:340) at org.jboss.invocation.InterceptorContext.run(InterceptorContext.java:356) at org.jboss.invocation.PrivilegedWithCombinerInterceptor.processInvocation(PrivilegedWithCombinerInterceptor.java:80) at org.jboss.invocation.InterceptorContext.proceed(InterceptorContext.java:340) at org.jboss.invocation.ChainedInterceptor.processInvocation(ChainedInterceptor.java:61) at org.jboss.as.ee.component.BasicComponent.constructComponentInstance(BasicComponent.java:161) at org.jboss.as.ee.component.BasicComponent.constructComponentInstance(BasicComponent.java:134) at org.jboss.as.ee.component.BasicComponent.createInstance(BasicComponent.java:88) at org.jboss.as.ejb3.component.singleton.SingletonComponent.getComponentInstance(SingletonComponent.java:124) at org.jboss.as.ejb3.component.singleton.SingletonComponent.start(SingletonComponent.java:138) at org.jboss.as.ee.component.ComponentStartService$1.run(ComponentStartService.java:54) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) at org.jboss.threads.JBossThread.run(JBossThread.java:320) 15:03:42,815 WARN [com.datastax.driver.core.Cluster] (cluster1-worker-1) Re-preparing already prepared query select DATA, VERSION from EVENTS where NAME = :name allow filtering. Please note that preparing the same query more than once is generally an anti-pattern and will likely affect performance. Consider preparing the statement only once. 15:03:43,221 INFO [org.apache.kafka.clients.consumer.internals.AbstractCoordinator] (EE-ManagedThreadFactory-default-Thread-2) Discovered coordinator contabo:9092 (id: 2147483647 rack: null) for group pokertracker. 15:03:43,223 INFO [org.apache.kafka.clients.consumer.internals.ConsumerCoordinator] (EE-ManagedThreadFactory-default-Thread-2) Revoking previously assigned partitions [] for group pokertracker 15:03:43,224 INFO [org.apache.kafka.clients.consumer.internals.AbstractCoordinator] (EE-ManagedThreadFactory-default-Thread-2) (Re-)joining group pokertracker 15:03:43,246 INFO [org.apache.kafka.clients.consumer.internals.AbstractCoordinator] (EE-ManagedThreadFactory-default-Thread-2) Successfully joined group pokertracker with generation 1 15:03:43,247 INFO [org.apache.kafka.clients.consumer.internals.ConsumerCoordinator] (EE-ManagedThreadFactory-default-Thread-2) Setting newly assigned partitions [test-0] for group pokertracker And here Is the log output from the docker container: 13:13:33,627 INFO [org.apache.kafka.clients.consumer.ConsumerConfig] (ServerService Thread Pool -- 61) ConsumerConfig values: metric.reporters = [] metadata.max.age.ms = 300000 partition.assignment.strategy = [org.apache.kafka.clients.consumer.RangeAssignor] reconnect.backoff.ms = 50 sasl.kerberos.ticket.renew.window.factor = 0.8 max.partition.fetch.bytes = 1048576 bootstrap.servers = [172.17.0.1:9092] ssl.keystore.type = JKS enable.auto.commit = true sasl.mechanism = GSSAPI interceptor.classes = null exclude.internal.topics = true ssl.truststore.password = null client.id = consumer-1 ssl.endpoint.identification.algorithm = null max.poll.records = 2147483647 check.crcs = true request.timeout.ms = 40000 heartbeat.interval.ms = 3000 auto.commit.interval.ms = 5000 receive.buffer.bytes = 65536 ssl.truststore.type = JKS ssl.truststore.location = null ssl.keystore.password = null fetch.min.bytes = 1 send.buffer.bytes = 131072 value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer group.id = pokertracker retry.backoff.ms = 100 sasl.kerberos.kinit.cmd = /usr/bin/kinit sasl.kerberos.service.name = null sasl.kerberos.ticket.renew.jitter = 0.05 ssl.trustmanager.algorithm = PKIX ssl.key.password = null fetch.max.wait.ms = 500 sasl.kerberos.min.time.before.relogin = 60000 connections.max.idle.ms = 540000 session.timeout.ms = 10000 metrics.num.samples = 2 key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer ssl.protocol = TLS ssl.provider = null ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1] ssl.keystore.location = null ssl.cipher.suites = null security.protocol = PLAINTEXT ssl.keymanager.algorithm = SunX509 metrics.sample.window.ms = 30000 auto.offset.reset = latest 13:13:33,650 INFO [org.apache.kafka.common.utils.AppInfoParser] (ServerService Thread Pool -- 61) Kafka version : 0.10.0.0 13:13:33,650 INFO [org.apache.kafka.common.utils.AppInfoParser] (ServerService Thread Pool -- 61) Kafka commitId : b8642491e78c5a13 13:13:33,711 WARN [com.datastax.driver.core.Cluster] (cluster1-worker-1) Re-preparing already prepared query select DATA, VERSION from EVENTS where NAME = :name allow filtering. Please note that preparing the same query more than once is generally an anti-pattern and will likely affect performance. Consider preparing the statement only once. 13:13:33,948 INFO [org.apache.kafka.clients.consumer.internals.AbstractCoordinator] (EE-ManagedThreadFactory-default-Thread-1) Discovered coordinator contabo:9092 (id: 2147483647 rack: null) for group pokertracker. 13:13:34,009 INFO [org.apache.kafka.clients.consumer.internals.AbstractCoordinator] (EE-ManagedThreadFactory-default-Thread-1) Marking the coordinator contabo:9092 (id: 2147483647 rack: null) dead for group pokertracker Here is the java code: @Startup @ConcurrencyManagement(ConcurrencyManagementType.BEAN) @Singleton public class InMemoryCache { @Inject KafkaConsumer<String, String> consumer; @Dedicated @Inject ExecutorService kafka; ... @PostConstruct public void onInit() { ... CompletableFuture .runAsync(this::handleKafkaEvent, kafka); } public void handleKafkaEvent() { while (true) { ConsumerRecords<String, String> records = consumer.poll(200); for (ConsumerRecord<String, String> record : records) { switch (record.topic()) { case KafkaProvider.TOPIC: System.out.println("record.value() = " + record.value()); List<CoreEvent> events = converter.convertToEvents(record.value()); for (CoreEvent event : events) { handle(event); } break; default: throw new IllegalArgumentException("Illegal message type: "); } } } } ...
