I am a novice in Apache camel and even more in camel-K. I am building a Kubernetes-containerized stack with an ActiveMQ message queue as inbound interface to external data sources, an InfluxDB as data store and a Camel-K route for message routing from AMQ to InfluxDB. Besides the Camel-K route, the system is running fine. I expose port 61616 of ActiveMQ to the cluster via a static-IP endpoint assigned to a service:
apiVersion: v1
kind: Service
metadata:
name: activemq-external
spec:
selector:
app: activemq
type: NodePort
ports:
- name: port-service-console
port: 8161
targetPort: 8161
nodePort: 8161
protocol: TCP
---
# Service definition for internal service with static IP (which can be used in camel-K integration "ActiveMQToInfluxDB.yaml")
# see https://kubernetes.io/docs/concepts/services-networking/service/#service-resource
apiVersion: v1
kind: Service
metadata:
name: activemq-internal
spec:
# type: NodePort
ports:
- name: port-internal
port: 61616
targetPort: 61616
# nodePort: 61616
protocol: TCP
---
apiVersion: v1
kind: Endpoints
metadata:
name: activemq-internal
subsets:
- addresses:
- ip: 172.17.0.8
ports:
- port: 61616
Hence I expect the port should be accessible to other pods in the cluster via 172.17.0.8:61616.
I am running the camel-K route by the command (never mind the dependency to apache commons, I need this for string manipulation)
kamel run kamel-integrations/ActiveMQToInfluxDbRoute.java -d mvn:org.apache.commons:commons-lang3:3.9 -d mvn:org.influxdb:influxdb-java:2.17 -d mvn:org.apache.activemq:activemq-camel:5.15.11 -d mvn:org.apache.camel:camel-core:2.25.0
The java code of the integration looks as follows:
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.camel.BindToRegistry;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.Message;
import org.apache.commons.lang3.StringUtils;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
//import org.springframework.context.annotation.Bean;
//import org.springframework.context.annotation.Configuration;
public class ActiveMQToInfluxDbRoute extends RouteBuilder {
private static final String MESSAGE_HEADER = "InfluxTimestamp";
@BindToRegistry
public ActiveMQConnectionFactory registerActiveMQConnectionFactory() {
System.out.println("ActiveMQ Listener: STARTING...");
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
connectionFactory.setBrokerURL("tcp://172.17.0.8:61616");
connectionFactory.setUserName("admin");
connectionFactory.setPassword("admin");
connectionFactory.setUseAsyncSend(false);
connectionFactory.setClientID("Influx Message Queue");
connectionFactory.setConnectResponseTimeout(300);
System.out.println("ActiveMQ Listener: STARTED");
return connectionFactory;
}
@Override
public void configure() throws Exception {
String sourceString = "activemq:queue:cryring_db_inbound?brokerURL=tcp://172.17.0.8:61616";
String targetString = "influxdb://influxDb?databaseName=my_database=true&retentionPolicy=default";
from(sourceString) //
.process(messagePayload -> {
String manipulation stuff
})//
.to(targetString) //
.onException(Exception.class) //
.useOriginalMessage() //
.handled(true) //
.log("error") //
.to("stream:out");
}
The execution of the java code fails due with the exception:
Route(route1)[From[activemq:queue:my_database?brokerU... because of Failed to resolve endpoint: activemq://queue:my_database?brokerURL=tcp%3A%2F%2F172.17.0.8%3A61616 due to: java.lang.IllegalArgumentException: wrong number of arguments
I found already this post with a similar issue: https://issues.apache.org/jira/browse/CAMEL-13145 but I cannot figure out what I have to do.
