Camel: detect Kafka not available
I am trying to detect when Kafka is not available. I have modified the example - https://github.com/apache/camel/blob/master/examples/camel-example-kafka/src/main/java/org/apache/camel/example/kafka/MessageConsumerClient.java and added following code right after camelContext.start() final Collection endpoints = camelContext.getEndpoints(); for (Endpoint endpoint : endpoints) { if (endpoint instanceof DefaultEndpoint) { final DefaultEndpoint endpoint1 = (DefaultEndpoint) endpoint; endpoint1.setBridgeErrorHandler(true); final HashMap consumerProperties = new HashMap<>(); consumerProperties.put("backoffMultiplier", 10); consumerProperties.put("backoffErrorThreshold", 5); endpoint1.setConsumerProperties(consumerProperties); } } I ran the main() and hoped to see the consumer stopping the attempts to connect to Kafka after 5 tries, 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? Thanks.
Re: Camel: detect Kafka not available
But this is exactly my point - there is no stack trace! I want there to be a stacktrace so that I could catch it using the regular error handler. This is the purpose of the call to endpoint1.setBridgeErrorHandler(true); However, the call does not work, no trace is printed and the polling continues. Here is the complete code based on https://github.com/apache/ camel/blob/master/examples/camel-example-kafka/src/main/ java/org/apache/camel/example/kafka/MessageConsumerClient.java with my addition between the *// change start *and *// change end *blocks: package org.apache.camel.example.kafka; import org.apache.camel.CamelContext; import org.apache.camel.Endpoint; import org.apache.camel.Exchange; import org.apache.camel.Processor; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.kafka.KafkaEndpoint; import org.apache.camel.component.properties.PropertiesComponent; import org.apache.camel.impl.DefaultCamelContext; import org.apache.camel.impl.DefaultEndpoint; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Collection; import java.util.HashMap; import java.util.Iterator; public final class MessageConsumerClient { private static final Logger LOG = LoggerFactory.getLogger(MessageConsumerClient.class); private MessageConsumerClient() { } public static void main(String[] args) throws Exception { LOG.info("About to run Kafka-camel integration..."); CamelContext camelContext = new DefaultCamelContext(); // Add route to send messages to Kafka camelContext.addRoutes(new RouteBuilder() { public void configure() { PropertiesComponent pc = getContext().getComponent("properties", PropertiesComponent.class); pc.setLocation("classpath:application.properties"); log.info("About to start route: Kafka Server -> Log "); onException(Exception.class).process(new Processor() { @Override public void process(Exchange exchange) throws Exception { System.out.println("Exception occurred!!"); } }); from("kafka:{{consumer.topic}}?brokers={{kafka.host}}:{{kafka.port}}" + "&maxPollRecords={{consumer.maxPollRecords}}" + "&consumersCount={{consumer.consumersCount}}" + "&seekTo={{consumer.seekTo}}" + "&groupId={{consumer.group}}") .routeId("FromKafka") .log("${body}"); } }); camelContext.start(); // // change start // final Collection endpoints = camelContext.getEndpoints(); for (Endpoint endpoint : endpoints) { if (endpoint instanceof DefaultEndpoint) { final DefaultEndpoint endpoint1 = (DefaultEndpoint) endpoint; endpoint1.setBridgeErrorHandler(true); final HashMap consumerProperties = new HashMap<>(); consumerProperties.put("backoffMultiplier", 10); consumerProperties.put("backoffErrorThreshold", 5); endpoint1.setConsumerProperties(consumerProperties); } } // // change end here // // let it run for 5 minutes before shutting down Thread.sleep(5 * 60 * 1000); camelContext.stop(); } } 2017-11-25 19:18 GMT+02:00 Claus Ibsen : > Post the stactrace so we can see from where the error is thrown. > > > On Wed, Nov 22, 2017 at 11:01 AM, Yacov Schondorf > wrote: > > I am trying to detect when Kafka is not available. I have modified the > > example - > > > > https://github.com/apache/camel/blob/master/examples/ > camel-example-kafka/src/main/java/org/apache/camel/example/ > kafka/MessageConsumerClient.java > > and > > added following code right after camelContext.start() > > > > > > > > final Collection endpoints = > camelContext.getEndpoints(); > > > > for (Endpoint endpoint : endpoints) { > > > > if (endpoint instanceof DefaultEndpoint) { > > > > final DefaultEndpoint endpoint1 = (DefaultEndpoint) > > endpoint; > > > > endpoint1.setBridgeErrorHandler(true); > > > > final HashMap consumerProperties = new > > HashMap<>(); > > > > consumerProperties.put("backoffMultiplier", 10); > > > > consumerProperties.put("backoffErrorThreshold", 5); > > > > endpoint1.setConsumerPrope
Re: Camel: detect Kafka not available
No solution by Camel for detecting connection errors? I gave a very clear reproducible scenario... 2017-11-28 11:44 GMT+02:00 Yacov Schondorf : > But this is exactly my point - there is no stack trace! I want there to be > a stacktrace so that I could catch it using the regular error handler. This > is the purpose of the call to endpoint1.setBridgeErrorHandler(true); > However, the call does not work, no trace is printed and the polling > continues. Here is the complete code based on https://github.com/apache/c > amel/blob/master/examples/camel-example-kafka/src/main/java/ > org/apache/camel/example/kafka/MessageConsumerClient.java with my > addition between the *// change start *and *// change end *blocks: > > package org.apache.camel.example.kafka; > > import org.apache.camel.CamelContext; > import org.apache.camel.Endpoint; > import org.apache.camel.Exchange; > import org.apache.camel.Processor; > import org.apache.camel.builder.RouteBuilder; > import org.apache.camel.component.kafka.KafkaEndpoint; > import org.apache.camel.component.properties.PropertiesComponent; > import org.apache.camel.impl.DefaultCamelContext; > import org.apache.camel.impl.DefaultEndpoint; > import org.slf4j.Logger; > import org.slf4j.LoggerFactory; > > import java.util.Collection; > import java.util.HashMap; > import java.util.Iterator; > > public final class MessageConsumerClient { > > private static final Logger LOG = LoggerFactory.getLogger( > MessageConsumerClient.class); > > private MessageConsumerClient() { > } > > public static void main(String[] args) throws Exception { > > LOG.info("About to run Kafka-camel integration..."); > > CamelContext camelContext = new DefaultCamelContext(); > > // Add route to send messages to Kafka > > camelContext.addRoutes(new RouteBuilder() { > public void configure() { > PropertiesComponent pc = > getContext().getComponent("properties", > PropertiesComponent.class); > pc.setLocation("classpath:application.properties"); > > log.info("About to start route: Kafka Server -> Log "); > onException(Exception.class).process(new Processor() { > @Override > public void process(Exchange exchange) throws > Exception { > System.out.println("Exception occurred!!"); > } > }); > from("kafka:{{consumer.topic}}?brokers={{kafka.host}}:{{ > kafka.port}}" > + "&maxPollRecords={{consumer.maxPollRecords}}" > + "&consumersCount={{consumer.consumersCount}}" > + "&seekTo={{consumer.seekTo}}" > + "&groupId={{consumer.group}}") > .routeId("FromKafka") > .log("${body}"); > } > }); > > camelContext.start(); > > // > // change start > // > final Collection endpoints = camelContext.getEndpoints(); > for (Endpoint endpoint : endpoints) { > if (endpoint instanceof DefaultEndpoint) { > final DefaultEndpoint endpoint1 = (DefaultEndpoint) > endpoint; > endpoint1.setBridgeErrorHandler(true); > final HashMap consumerProperties = new > HashMap<>(); > consumerProperties.put("backoffMultiplier", 10); > consumerProperties.put("backoffErrorThreshold", 5); > endpoint1.setConsumerProperties(consumerProperties); > } > } > // > // change end here > // > > // let it run for 5 minutes before shutting down > Thread.sleep(5 * 60 * 1000); > > camelContext.stop(); > } > > } > > > > > > 2017-11-25 19:18 GMT+02:00 Claus Ibsen : > >> Post the stactrace so we can see from where the error is thrown. >> >> >> On Wed, Nov 22, 2017 at 11:01 AM, Yacov Schondorf >> wrote: >> > I am trying to detect when Kafka is not available. I have modified the >> > example - >> > >> > https://github.com/apache/camel/blob/master/examples/camel- >> example-kafka/src/main/java/org/apache/camel/example/kafka >> /MessageConsumerClient.java >> > and >> > added following code right after camelContext.start() >> > >> > >> > >> >
Re: Camel: detect Kafka not available
Just for the record, I have tried adding "bridgeErrorHandler=true" to the route . This does not help and Camel still tries to connect to the non-existing Kafka. Final route looks like this: from("kafka:{{consumer.topic}}?bridgeErrorHandler=true&brokers={{kafka.host}}:{{kafka.port}}" + "&maxPollRecords={{consumer.maxPollRecords}}" + "&consumersCount={{consumer.consumersCount}}" + "&seekTo={{consumer.seekTo}}" + "&groupId={{consumer.group}}") I was expecting an exception to be thrown but this does not happen. This is based on https://github.com/apache/camel/blob/master/examples/camel-e xample-kafka/src/main/java/org/apache/camel/example/kafka/ MessageConsumerClient.java with just an additional "bridgeErrorHandler=true" added to the route. 2017-12-03 14:54 GMT+02:00 Yacov Schondorf : > No solution by Camel for detecting connection errors? I gave a very clear > reproducible scenario... > > 2017-11-28 11:44 GMT+02:00 Yacov Schondorf : > >> But this is exactly my point - there is no stack trace! I want there to >> be a stacktrace so that I could catch it using the regular error handler. >> This is the purpose of the call to endpoint1.setBridgeErrorHandler(true); >> However, the call does not work, no trace is printed and the polling >> continues. Here is the complete code based on https://github.com/apache/c >> amel/blob/master/examples/camel-example-kafka/src/main/java/ >> org/apache/camel/example/kafka/MessageConsumerClient.java with my >> addition between the *// change start *and *// change end *blocks: >> >> package org.apache.camel.example.kafka; >> >> import org.apache.camel.CamelContext; >> import org.apache.camel.Endpoint; >> import org.apache.camel.Exchange; >> import org.apache.camel.Processor; >> import org.apache.camel.builder.RouteBuilder; >> import org.apache.camel.component.kafka.KafkaEndpoint; >> import org.apache.camel.component.properties.PropertiesComponent; >> import org.apache.camel.impl.DefaultCamelContext; >> import org.apache.camel.impl.DefaultEndpoint; >> import org.slf4j.Logger; >> import org.slf4j.LoggerFactory; >> >> import java.util.Collection; >> import java.util.HashMap; >> import java.util.Iterator; >> >> public final class MessageConsumerClient { >> >> private static final Logger LOG = LoggerFactory.getLogger(Messag >> eConsumerClient.class); >> >> private MessageConsumerClient() { >> } >> >> public static void main(String[] args) throws Exception { >> >> LOG.info("About to run Kafka-camel integration..."); >> >> CamelContext camelContext = new DefaultCamelContext(); >> >> // Add route to send messages to Kafka >> >> camelContext.addRoutes(new RouteBuilder() { >> public void configure() { >> PropertiesComponent pc = >> getContext().getComponent("properties", >> PropertiesComponent.class); >> pc.setLocation("classpath:application.properties"); >> >> log.info("About to start route: Kafka Server -> Log "); >> onException(Exception.class).process(new Processor() { >> @Override >> public void process(Exchange exchange) throws >> Exception { >> System.out.println("Exception occurred!!"); >> } >> }); >> from("kafka:{{consumer.topic}} >> ?brokers={{kafka.host}}:{{kafka.port}}" >> + "&maxPollRecords={{consumer.maxPollRecords}}" >> + "&consumersCount={{consumer.consumersCount}}" >> + "&seekTo={{consumer.seekTo}}" >> + "&groupId={{consumer.group}}") >> .routeId("FromKafka") >> .log("${body}"); >> } >> }); >> >> camelContext.start(); >> >> // >> // change start >> // >> final Collection endpoints = >> camelContext.getEndpoints(); >> for (Endpoint endpoint : endpoints) { >> if (endpoint instanceof DefaultEndpoint) { >> final DefaultEndpoint endpoint1 = (DefaultEndpoint) >> endpoint; >> end
mocking future routes
Hi, I know how to mock/advise existing routes. But what do I do if one of my routes creates new routes? For example, I can send a configuration message to a configuration route which has an activeMQ From endpoint. The endpoint is advised and replaced with direct:a, so I send a configuration message in my test. The route continues by creating a new route from activeMQ:someTopic to kafka:someTopic. This route fails to create because there is no real ssl-enabled kafka and activeMQ in the test environment. Is there anything that can be done?
Unable to view message body
Hi, I am trying to log the contents of a message using the log component. Following is my route: from("jetty://http://localhost:8099/myApi ").noStreamCaching().to("log:myCategory?level=DEBUG&showBody=true&showHeaders=true"); Although I used noStreamCaching() in the route, and specified showBody=true, I do not see the message body. Instead, I see: Body: [Body is instance of org.apache.camel.StreamCache] I run following commands from Firefox development console in order to test the route: var script = document.createElement('script'); script.type = 'text/javascript'; script.src = ' https://ajax.googleapis.com/ajax/libs/jquery/3.3.1/jquery.min.js' document.head.appendChild(script); $.ajax({ url: "http://localhost:8099/myApi";, headers: { "Content-Type":"application/json" }, method: "POST", dataType: "json", data: "{\ 'foo': 'bar'\ }" }) Any ideas what I am doing wrong?
exceptionMessage() not working properly
I have the following global onException clause in my RouteBuilder configure method: onException(BadHeaderException.class) .handled(true) .transform(constant("Bad API request: " + exceptionMessage())) .setHeader(Exchange.HTTP_RESPONSE_CODE, constant(400)) The route is as follows: from(String.format("jetty://http://0.0.0.0:%d%s",port, KEEP_ALIVE_URI)) .process(new MyRequiredHeaderProcessor()) .log("My API called") .setBody(simple(KEEP_ALIVE_SUCCESS_MESSAGE)); and MyRequiredHeaderProcessor may throw BadHeaderException: class MyRequiredHeaderProcessor implements Processor { @Override public void process(Exchange exchange) throws Exception { final Object myHeader = exchange.getIn().getHeader("MyRequiredHeader"); if (myHeader == null) { throw new BadHeaderException ("Missing header"); } if (!myHeader .equals(someValue)) { throw new BadHeaderException("Incorrect header value"); } } } When I call the jetty URL without the required header, the request is rejected with error code 400 - as expected - BUT the error message is: Bad API request: simple{${exception.message}} Shouldn't exceptionMessage() return the text I set in MyRequiredHeaderProcessor?
Re: exceptionMessage() not working properly
Thank you! This worked for me: onException(BadHeaderException.class) .handled(true) * .transform(simple("Bad API request: ${exception.message}"))* .setHeader(Exchange.HTTP_RESPONSE_CODE, constant(400)) 2018-02-02 11:33 GMT+02:00 Claus Ibsen : > Hi > > The configure method in the RouteBuilder is only invoked once to setup > the routes. > And constant is creating a constant message. > > So what you need to do is to use simple instead to tell Camel to build > a dynamic message with the exception message > > See the docs at: http://camel.apache.org/simple > > > > On Thu, Feb 1, 2018 at 8:43 PM, Yacov Schondorf > wrote: > > I have the following global onException clause in my RouteBuilder > configure > > method: > > > > onException(BadHeaderException.class) > > .handled(true) > > .transform(constant("Bad API request: " + exceptionMessage())) > > .setHeader(Exchange.HTTP_RESPONSE_CODE, constant(400)) > > > > The route is as follows: > > > > from(String.format("jetty://http://0.0.0.0:%d%s",port, KEEP_ALIVE_URI)) > > .process(new MyRequiredHeaderProcessor()) > > .log("My API called") > > .setBody(simple(KEEP_ALIVE_SUCCESS_MESSAGE)); > > > > and MyRequiredHeaderProcessor may throw BadHeaderException: > > > > class MyRequiredHeaderProcessor implements Processor { > > @Override > > public void process(Exchange exchange) throws Exception { > > final Object myHeader = exchange.getIn().getHeader("MyRequiredHeader"); > > if (myHeader == null) { > > throw new BadHeaderException ("Missing header"); > > } > > if (!myHeader .equals(someValue)) { > > throw new BadHeaderException("Incorrect header value"); > > } > > } > > } > > > > When I call the jetty URL without the required header, the request is > > rejected with error code 400 - as expected - BUT the error message is: > > > > Bad API request: simple{${exception.message}} > > > > Shouldn't exceptionMessage() return the text I set in > > MyRequiredHeaderProcessor? > > > > -- > Claus Ibsen > - > http://davsclaus.com @davsclaus > Camel in Action 2: https://www.manning.com/ibsen2 > -- יעקב שיינדורף מנכ"ל ומנהל פיתוח, מילות מפתח לאתרים 054-5750201 http://keywords-4-websites.com <http://keywords-4-websites.com/home.jsp> יעקב שיינדורף מנהל מוסיקלי StyleBrass 054-5750201 http://www.stylebrass.co.il
best approach to perform client certificate validation and restrict access to single IP
Hello, I have a requirement to provide a service using Camel Jetty component and SSL. I have the SSL working fine for server certificate validation by clients. The route basically looks like this: from("jetty://https://0.0.0.0:thePort/theSecureAPI";).to(...) The code that configures the SSL for the Jetty component looks like below and works fine: private void configureJettyComponentForSsl() { KeyStoreParameters ksp = new KeyStoreParameters(); ksp.setResource(trustStorePath); ksp.setPassword(trustStorePassword); KeyManagersParameters kmp = new KeyManagersParameters(); kmp.setKeyStore(ksp); kmp.setKeyPassword(keyPassword); SSLContextParameters scp = new SSLContextParameters(); scp.setKeyManagers(kmp); JettyHttpComponent jettyComponent = getContext().getComponent("jetty", JettyHttpComponent.class); jettyComponent.setSslContextParameters(scp); } Now I need to add client certificate validation, and to restrict connections to a particular IP. What would be the best approach? Maybe using SslSocketConnectors like: final HashMap portToConnectorMap = new HashMap<>(); portToConnectorMap.put(thePort, what-here?? ) jettyComponent.setSslSocketConnectors(portToConnectorMap); Or, maybe I could accept the call into the route and use a processor that would reject it if it does not come from the required IP? (how can I get the client IP from within the route?) I would still need to perfor the client certificate validation. Can I use a spring security filter? Any guidance would be welcome.
Re: mocking future routes
Hi, So sorry for the late response; only now I was able to get back to this issue. I don't see how the stub component can help me here. This is a real route I need to mock in a production code. Consider this use case: 1. My route (let's call it route 1) has a Jetty start endpoint, and is a configuration service so that an incoming call would cause a configuration change to the system. If the incoming call has some "flag" set to "true" then a new route needs to be created (route 2). Route 2 has a Kafka endpoint which I need to mock because the connection parameters are not relevant at test time. 2. On my test, I advise route 1, and replace the Jetty endpoint with a start endpoint, and send a message that should trigger creation of route 2. Creation of route 2 fails because of the wrong Kafka parameters. If I add "stub:" in route 2 ("Stub:kafka...") then in non-test time the route will not work. Any solution for this? 2018-02-01 18:58 GMT+02:00 Claus Ibsen : > You can try when you create the new route, to add "stub:" as the > prefix for your endpoints. Then you can use the stub component to mock > them. > http://camel.apache.org/stub > > On Wed, Jan 17, 2018 at 6:10 PM, Yacov Schondorf > wrote: > > Hi, > > I know how to mock/advise existing routes. But what do I do if one of my > > routes creates new routes? For example, I can send a configuration > message > > to a configuration route which has an activeMQ From endpoint. The > endpoint > > is advised and replaced with direct:a, so I send a configuration message > in > > my test. The route continues by creating a new route from > > activeMQ:someTopic to kafka:someTopic. This route fails to create because > > there is no real ssl-enabled kafka and activeMQ in the test environment. > Is > > there anything that can be done? > > > > -- > Claus Ibsen > - > http://davsclaus.com @davsclaus > Camel in Action 2: https://www.manning.com/ibsen2 > -- יעקב שיינדורף מנכ"ל ומנהל פיתוח, מילות מפתח לאתרים 054-5750201 http://keywords-4-websites.com <http://keywords-4-websites.com/home.jsp> יעקב שיינדורף מנהל מוסיקלי StyleBrass 054-5750201 http://www.stylebrass.co.il
encoding problem on Jetty consumer side
I have a route defined from a Jetty endpoint. I am trying to read a path parameter and print its contents: from("jetty:http://localhost:8099/my-service?httpMethodRestrict=post";) .process(exchange -> { String path = (String) exchange.getIn().getHeader("path"); System.out.println("got path: " + path); }); } I am calling this via curl on Windows passing Hebrew characters: curl http://localhost:8099/my-service -X post -d "path=שדגכ" The output I am getting is: path= I tried curl http://localhost:8099/my-service -X post -H "Content-Type: text/html; charset=UTF-8" -d "path:שדגכ" In this case the path is not added as a Header at all so the code above produces null. I am able to see the path in the debug window using exchange.getIn().getBody(String.class) but then i see that the result is again path: Any idea how I can get set a utf-8 encoding properly on Camel side and on curl side?