Can you enable DEBUG logs ? It'll be helpful to debug. -- Kamal
On Mon, Jan 9, 2017 at 5:37 AM, Gupta, Swati <swati.gu...@anz.com> wrote: > Hello All, > > Any help on this would be appreciated. > There seems to be no error. Does it look like a version issue? > > I have updated my pom.xml with the below: > <dependency> > <groupId>org.springframework.kafka</groupId> > <artifactId>spring-kafka</artifactId> > <version>1.1.2.BUILD-SNAPSHOT</version> > </dependency> > > <dependency> > <groupId>org.apache.camel</groupId> > <artifactId>camel-kafka</artifactId> > <version>2.17.0</version> > </dependency> > > <dependency> > <groupId>org.apache.kafka</groupId> > <artifactId>kafka-clients</artifactId> > <version>0.10.1.0</version> > </dependency> > <dependency> > <groupId>org.apache.kafka</groupId> > <artifactId>kafka_2.11</artifactId> > <version>0.10.1.0</version> > </dependency> > > <dependency> > <groupId>org.apache.camel</groupId> > <artifactId>camel-core</artifactId> > <version>2.17.0</version> > </dependency> > > Thanks & Regards > Swati > > -----Original Message----- > From: Gupta, Swati [mailto:swati.gu...@anz.com] > Sent: Friday, 6 January 2017 4:01 PM > To: users@kafka.apache.org > Subject: RE: Apache Kafka integration using Apache Camel > > Yes, the kafka console consumer displays the message correctly. > I also tested the same with a Java application, it works fine. There seems > to be an issue with Camel route trying to consume. > > There is no error in the console. But, the logs show as below: > kafka.KafkaCamelTestConsumer > Connected to the target VM, address: '127.0.0.1:65007', transport: > 'socket' > PID_IS_UNDEFINED: INFO DefaultCamelContext - Apache Camel 2.17.0 > (CamelContext: camel-1) is starting > PID_IS_UNDEFINED: INFO ManagedManagementStrategy - JMX is enabled > PID_IS_UNDEFINED: INFO DefaultTypeConverter - Loaded 183 type converters > PID_IS_UNDEFINED: INFO DefaultRuntimeEndpointRegistry - Runtime endpoint > registry is in extended mode gathering usage statistics of all incoming and > outgoing endpoints (cache limit: 1000) > PID_IS_UNDEFINED: INFO DefaultCamelContext - AllowUseOriginalMessage is > enabled. If access to the original message is not needed, then its > recommended to turn this option off as it may improve performance. > PID_IS_UNDEFINED: INFO DefaultCamelContext - StreamCaching is not in use. > If using streams then its recommended to enable stream caching. See more > details at http://camel.apache.org/stream-caching.html > PID_IS_UNDEFINED: INFO KafkaConsumer - Starting Kafka consumer > PID_IS_UNDEFINED: INFO ConsumerConfig - ConsumerConfig values: > auto.commit.interval.ms = 5000 > auto.offset.reset = earliest > bootstrap.servers = [localhost:9092] > check.crcs = true > client.id = > connections.max.idle.ms = 540000 > enable.auto.commit = true > exclude.internal.topics = true > fetch.max.bytes = 52428800 > fetch.max.wait.ms = 500 > fetch.min.bytes = 1024 > group.id = testing > heartbeat.interval.ms = 3000 > interceptor.classes = null > key.deserializer = class org.apache.kafka.common.serialization. > StringDeserializer > max.partition.fetch.bytes = 1048576 > max.poll.interval.ms = 300000 > max.poll.records = 500 > metadata.max.age.ms = 300000 > metric.reporters = [] > metrics.num.samples = 2 > metrics.sample.window.ms = 30000 > partition.assignment.strategy = [org.apache.kafka.clients. > consumer.RangeAssignor] > receive.buffer.bytes = 32768 > reconnect.backoff.ms = 50 > request.timeout.ms = 40000 > retry.backoff.ms = 100 > sasl.kerberos.kinit.cmd = /usr/bin/kinit > sasl.kerberos.min.time.before.relogin = 60000 > sasl.kerberos.service.name = null > sasl.kerberos.ticket.renew.jitter = 0.05 > sasl.kerberos.ticket.renew.window.factor = 0.8 > sasl.mechanism = GSSAPI > security.protocol = PLAINTEXT > send.buffer.bytes = 131072 > session.timeout.ms = 30000 > ssl.cipher.suites = null > ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1] > ssl.endpoint.identification.algorithm = null > ssl.key.password = null > ssl.keymanager.algorithm = SunX509 > ssl.keystore.location = null > ssl.keystore.password = null > ssl.keystore.type = JKS > ssl.protocol = TLS > ssl.provider = null > ssl.secure.random.implementation = null > ssl.trustmanager.algorithm = PKIX > ssl.truststore.location = null > ssl.truststore.password = null > ssl.truststore.type = JKS > value.deserializer = class org.apache.kafka.common.serialization. > StringDeserializer > > PID_IS_UNDEFINED: INFO ConsumerConfig - ConsumerConfig values: > auto.commit.interval.ms = 5000 > auto.offset.reset = earliest > bootstrap.servers = [localhost:9092] > check.crcs = true > client.id = consumer-1 > connections.max.idle.ms = 540000 > enable.auto.commit = true > exclude.internal.topics = true > fetch.max.bytes = 52428800 > fetch.max.wait.ms = 500 > fetch.min.bytes = 1024 > group.id = testing > heartbeat.interval.ms = 3000 > interceptor.classes = null > key.deserializer = class org.apache.kafka.common.serialization. > StringDeserializer > max.partition.fetch.bytes = 1048576 > max.poll.interval.ms = 300000 > max.poll.records = 500 > metadata.max.age.ms = 300000 > metric.reporters = [] > metrics.num.samples = 2 > metrics.sample.window.ms = 30000 > partition.assignment.strategy = [org.apache.kafka.clients. > consumer.RangeAssignor] > receive.buffer.bytes = 32768 > reconnect.backoff.ms = 50 > request.timeout.ms = 40000 > retry.backoff.ms = 100 > sasl.kerberos.kinit.cmd = /usr/bin/kinit > sasl.kerberos.min.time.before.relogin = 60000 > sasl.kerberos.service.name = null > sasl.kerberos.ticket.renew.jitter = 0.05 > sasl.kerberos.ticket.renew.window.factor = 0.8 > sasl.mechanism = GSSAPI > security.protocol = PLAINTEXT > send.buffer.bytes = 131072 > session.timeout.ms = 30000 > ssl.cipher.suites = null > ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1] > ssl.endpoint.identification.algorithm = null > ssl.key.password = null > ssl.keymanager.algorithm = SunX509 > ssl.keystore.location = null > ssl.keystore.password = null > ssl.keystore.type = JKS > ssl.protocol = TLS > ssl.provider = null > ssl.secure.random.implementation = null > ssl.trustmanager.algorithm = PKIX > ssl.truststore.location = null > ssl.truststore.password = null > ssl.truststore.type = JKS > value.deserializer = class org.apache.kafka.common.serialization. > StringDeserializer > > PID_IS_UNDEFINED: INFO AppInfoParser - Kafka version : 0.10.1.0 > PID_IS_UNDEFINED: INFO AppInfoParser - Kafka commitId : 3402a74efb23d1d4 > PID_IS_UNDEFINED: INFO DefaultCamelContext - Route: route1 started and > consuming from: Endpoint[kafka://localhost:9092?autoOffsetReset=earliest& > consumersCount=1&groupId=testing&topic=test] > PID_IS_UNDEFINED: INFO DefaultCamelContext - Total 1 routes, of which 1 > are started. > PID_IS_UNDEFINED: INFO DefaultCamelContext > > -----Original Message----- > From: Ewen Cheslack-Postava [mailto:e...@confluent.io] > Sent: Friday, 6 January 2017 3:58 PM > To: users@kafka.apache.org > Subject: Re: Apache Kafka integration using Apache Camel > > More generally, do you have any log errors/messages or additional info? > It's tough to debug issues like this from 3rd party libraries if they > don't provide logs/exception info that indicates why processing a specific > message failed. > > -Ewen > > On Thu, Jan 5, 2017 at 8:29 PM, UMESH CHAUDHARY <umesh9...@gmail.com> > wrote: > > > Did you test that kafka console consumer is displaying the produced > > message? > > > > On Fri, Jan 6, 2017 at 9:18 AM, Gupta, Swati <swati.gu...@anz.com> > wrote: > > > > > Hello All, > > > > > > > > > > > > I am trying to create a Consumer using Apache Camel for a topic in > > > Apache Kafka. > > > I am using Camel 2.17.0 and Kafka 0.10 and JDK 1.8. > > > I have attached a file, KafkaCamelTestConsumer.java which is a > > > standalone application trying to read from a topic “test1”created > > > in Apache Kafka I am producing messages from the console and also > > > was successful to produce messages using a Camel program in the > > > topic "test1", but not able to consume messages. Ideally, it should > > > get printed, but nothing seems to happen. The log says that the > > > route has started but does not process any message. > > > > > > Please help to confirm if there is anything wrong with the below > syntax: > > > > > > from(*"kafka:localhost:9092?topic=test1&groupId=testingGroupNew& > > autoOffsetReset=earliest" > > > *+ > > > > > > *"&consumersCount=1&keyDeserializer=org.apache. > > kafka.common.serialization.StringDeserializer&" > > > *+ > > > *"valueDeserializer=org.apache.kafka.common.serialization. > > StringDeserializer" > > > *+ > > > *"&autoCommitIntervalMs=1000&sessionTimeoutMs=30000& > > autoCommitEnable=true"* > > > ).split() > > > .body() > > > .process(*new *Processor() { > > > @Override > > > *public void *process(Exchange exchange) > > > *throws *Exception { > > > String messageKey = *""*; > > > *if *(exchange.getIn() != *null*) { > > > Message message = exchange.getIn(); > > > Integer partitionId = (Integer) message > > > .getHeader(KafkaConstants.* > > PARTITION* > > > ); > > > String topicName = (String) message > > > .getHeader(KafkaConstants.* > TOPIC*); > > > *if *(message.getHeader( > > KafkaConstants.*KEY*) > > > != *null*) > > > messageKey = (String) message > > > .getHeader(KafkaConstants.* > > KEY*); > > > Object data = message.getBody(); > > > > > > > > > System.*out*.println( > > > *"topicName :: " *+ topicName + > > > *" partitionId :: " *+ partitionId > + > > > *" messageKey :: " *+ messageKey + > > > *" message :: " *+ data + > > *"**\n**"*); > > > } > > > } > > > }).to( > > > *"file://C:/swati/?fileName=MyOutputFile.txt&charset=utf-8"*); > > > } > > > }); > > > > > > > > > > > > I have also tried with the basic parameters as below and it still > > > fails > > to > > > read messages. > > > > > > from( > > > *"kafka:localhost:9092?topic=test1&groupId=testingGroupNew& > > autoOffsetReset=earliest")* > > > > > > Any help on this will be greatly appreciated. > > > > > > Thanks in advance > > > > > > > > > > > > Thanks & Regards > > > > > > Swati > > > > > > ------------------------------ > > > This e-mail and any attachments to it (the "Communication") is, > > > unless otherwise stated, confidential, may contain copyright > > > material and is for the use only of the intended recipient. If you > > > receive the Communication > > in > > > error, please notify the sender immediately by return e-mail, delete > > > the Communication and the return e-mail, and do not read, copy, > > > retransmit or otherwise deal with it. Any views expressed in the > > > Communication are > > those > > > of the individual sender only, unless expressly stated to be those > > > of Australia and New Zealand Banking Group Limited ABN 11 005 357 > > > 522, or > > any > > > of its related entities including ANZ Bank New Zealand Limited > > > (together "ANZ"). ANZ does not accept liability in connection with > > > the integrity of or errors in the Communication, computer virus, > > > data corruption, interference or delay arising from or in respect of > the Communication. > > > > > > > > > > > This e-mail and any attachments to it (the "Communication") is, unless > otherwise stated, confidential, may contain copyright material and is for > the use only of the intended recipient. If you receive the Communication in > error, please notify the sender immediately by return e-mail, delete the > Communication and the return e-mail, and do not read, copy, retransmit or > otherwise deal with it. Any views expressed in the Communication are those > of the individual sender only, unless expressly stated to be those of > Australia and New Zealand Banking Group Limited ABN 11 005 357 522, or any > of its related entities including ANZ Bank New Zealand Limited (together > "ANZ"). ANZ does not accept liability in connection with the integrity of > or errors in the Communication, computer virus, data corruption, > interference or delay arising from or in respect of the Communication. > > > This e-mail and any attachments to it (the "Communication") is, unless > otherwise stated, confidential, may contain copyright material and is for > the use only of the intended recipient. If you receive the Communication in > error, please notify the sender immediately by return e-mail, delete the > Communication and the return e-mail, and do not read, copy, retransmit or > otherwise deal with it. Any views expressed in the Communication are those > of the individual sender only, unless expressly stated to be those of > Australia and New Zealand Banking Group Limited ABN 11 005 357 522, or any > of its related entities including ANZ Bank New Zealand Limited (together > "ANZ"). ANZ does not accept liability in connection with the integrity of > or errors in the Communication, computer virus, data corruption, > interference or delay arising from or in respect of the Communication. >
