I believe a response to this message was sent but it was blocked by our IT. Could you send the response again and also to yacov.schond...@gmail.com ? Thanks.
-----Original Message----- From: Schondorf, Yacov Sent: Tuesday, November 21, 2017 3:40 PM To: 'claus.ib...@gmail.com' <claus.ib...@gmail.com> Cc: 'users@camel.apache.org' <users@camel.apache.org> Subject: RE: EXTERNAL: Re: Re: Camel: detect Kafka wrong IP We have purchased the book! Section 11.5.8 talks about "Bridging the consumer with Camel’s error handler". The example uses XML and I found no example using Java DSL. Since our project uses Kafka and Java DSL, I improvised, and added to https://github.com/apache/camel/blob/master/examples/camel-example-kafka/src/main/java/org/apache/camel/example/kafka/MessageConsumerClient.java following code right after camelContext.start() final Collection<Endpoint> endpoints = camelContext.getEndpoints(); for (Endpoint endpoint : endpoints) { if (endpoint instanceof DefaultEndpoint) { final DefaultEndpoint endpoint1 = (DefaultEndpoint) endpoint; endpoint1.setBridgeErrorHandler(true); final HashMap<String, Object> consumerProperties = new HashMap<>(); consumerProperties.put("backoffMultiplier", 10); consumerProperties.put("backoffErrorThreshold", 5); endpoint1.setConsumerProperties(consumerProperties); } } I also added sone exception handling code: onException(Exception.class).process(new Processor() { @Override public void process(Exchange exchange) throws Exception { System.out.println("Exception occurred!!"); } }); I ran the main() and hoped to see the consumer stopping the attempts to connect to Kafka after 5 tries, and my exception handling code called, but this did not work. I keep getting output messages of “Connection to node -1 could not be established. Broker may not be available.” Is this the right way to go? What am I doing wrong? -----Original Message----- From: Schondorf, Yacov Sent: Thursday, November 16, 2017 4:55 PM To: users@camel.apache.org Subject: RE: EXTERNAL: Re: Re: Camel: detect Kafka wrong IP Will ask my employee to buy the book. I have studied the error handling page but found nothing that would help me resolve this issue. -----Original Message----- From: Claus Ibsen [mailto:claus.ib...@gmail.com] Sent: Thursday, November 16, 2017 4:38 PM To: users@camel.apache.org Subject: EXTERNAL: Re: Re: Camel: detect Kafka wrong IP On Thu, Nov 16, 2017 at 3:33 PM, Schondorf, Yacov <yschond...@forcepoint.com> wrote: > Hi, > > Thanks for your reply! I don't have the CIA2 book (would love to buy it > though). Is there an example online? > You can ask your employer to buy it - knowledge is cheap when you have it all in a book. The book has source code with a bunch of examples. Also you can study to Camel error handling pages on the website. > -----Original Message----- > From: Claus Ibsen [mailto:claus.ib...@gmail.com] > Sent: Thursday, November 16, 2017 4:28 PM > To: users@camel.apache.org > Subject: EXTERNAL: Re: Camel: detect Kafka wrong IP > > Hi > > This is expected. onException in the route builder only happens when you have > a Camel exchange/message to route. And you dont have that because you cannot > connect to Kafka. > > See for example the CiA2 book error handling chapter which talks about this, > and how some Camel components allow to bridge and report this as a Camel > exchange with the exception so you can deal with it using onException. > > On Thu, Nov 16, 2017 at 3:13 PM, Schondorf, Yacov <yschond...@forcepoint.com> > wrote: >> I am configuring Kafka as a source in my RouteBuilder. My goal is to handle >> Kafka disconnection issues. My RouteBuilder is as follows: >> new RouteBuilder() { >> public void configure() { >> onException(Exception.class).process(exchange -> { >> final Exception exception = exchange.getException(); >> logger.error(exception.getMessage()); >> // will do more processing here >> }); >> from(String.format("kafka:%s?brokers=%s:%s", topicName, host, >> port)).bean(getMyService(), "myMethod") >> .process(new Processor() { >> @Override >> public void process(Exchange exchange) throws Exception { >> // some more processing >> } >> }); >> } >> }; >> >> I am configuring Kafka as a source in my RouteBuilder. My goal is to handle >> Kafka disconnection issues. My RouteBuilder is as follows: >> >> new RouteBuilder() { >> >> public void configure() { >> >> onException(Exception.class).process(exchange -> { >> >> final Exception exception = exchange.getException(); >> >> logger.error(exception.getMessage()); >> >> // will do more processing here >> >> }); >> >> from(String.format("kafka:%s?brokers=%s:%s", topicName, >> host, port)).bean(getMyService(), "myMethod") >> >> .process(new Processor() { >> >> @Override >> >> public void process(Exchange exchange) throws >> Exception { >> >> // some more processing >> >> } >> >> }); >> >> } >> >> }; >> >> I provided wrong host and port, and expected to see an exception. However, >> no exception is seen in the log, and the onException processing is not get >> called. Any idea what I am doing wrong? >> >> A similar problem can be reproduced by running >> https://github.com/apache/camel/blob/master/examples/camel-example-kafka/src/main/java/org/apache/camel/example/kafka/MessageConsumerClient.java >> locally without any Kafka server running. Doing so results in a constant >> flow of messages: >> >> Connection to node -1 could not be established. Broker may not be available. >> Is there a way to have an exception thrown? Any help would be appreciated. >> >> YACOV SCHONDORF >> Sr. Software Engineer >> >> FORCEPOINT >> ph: +972.9.776.4233 >> fax: >> www.forcepoint.com<http://www.forcepoint.com> >> >> FORWARD WITHOUT FEAR >> >> >> > > > > -- > Claus Ibsen > ----------------- > http://davsclaus.com @davsclaus > Camel in Action 2: https://www.manning.com/ibsen2 -- Claus Ibsen ----------------- http://davsclaus.com @davsclaus Camel in Action 2: https://www.manning.com/ibsen2
