I am planning to pull the data from Hono using Strimzi kafka(Version 0.20.0) connector (Camel AMQP source connector). I have followed the below steps to read data from Hono.
I downloaded Camel-amqp-kafka-connector, JMS Jar files from below link:
https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-amqp-kafka-connector/0.7.0/camel-amqp-kafka-connector-0.7.0-package.tar.gz
https://downloads.apache.org/qpid/jms/0.51.0/apache-qpid-jms-0.51.0-bin.tar.gz
After downloaded the above tar and unzipped and created docker image file using below command
cat <Dockerfile
FROM strimzi/kafka:0.20.1-kafka-2.6.0
USER root:root
RUN mkdir -p /opt/kafka/plugins/camel
COPY ./camel-activemq-kafka-connector/* /opt/kafka/plugins/camel/
USER 1001
EOF
Docker build -f ./Dockerfile -t localhost:5000/my-connector-amqp_new .
docker push localhost:5000/my-connector-amqp_new
using below command i have created kafkaConnect( here used my local image created above)
apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaConnect
metadata:
name: my-connect-cluster
annotations:
strimzi.io/use-connector-resources: "true"
spec:
image: 10.128.0.6:5000/my-connector-amqp_new
replicas: 1
bootstrapServers: my-cluster-kafka-bootstrap:9092
config:
group.id: connect-cluster
offset.storage.topic: connect-cluster-offsets
config.storage.topic: connect-cluster-configs
status.storage.topic: connect-cluster-status
config.storage.replication.factor: 1
offset.storage.replication.factor: 1
status.storage.replication.factor: 1
**I was referring the below link to create kafkaConnector properties:
https://github.com/apache/camel-kafka-connector/blob/master/examples/CamelAmqpSourceConnector.properties**
Values are in this file as mentioned it below:
name=CamelAmqpSourceConnector
topics=mytopic
tasks.max=1
connector.class=org.apache.camel.kafkaconnector.amqp.CamelAmqpSourceConnector
camel.source.path.destinationType=queue
camel.source.path.destinationName=test-queue
camel.component.amqp.includeAmqpAnnotations=true
camel.component.amqp.connectionFactory=#class:org.apache.qpid.jms.JmsConnectionFactory
camel.component.amqp.connectionFactory.remoteURI=amqp://localhost:5672
camel.component.amqp.username=admin
camel.component.amqp.password=admin
camel.component.amqp.testConnectionOnStartup=true
I am using the below configurations. Could you suggest me what is the correct value for this property?
camel.component.amqp.connectionFactory
apiVersion: kafka.strimzi.io/v1alpha1
kind: KafkaConnector
metadata:
name: camelamqpsourceconnector
labels:
strimzi.io/cluster: my-connect-cluster-new
spec:
class: org.apache.camel.kafkaconnector.amqp.CamelAmqpSourceConnector
tasksMax: 1
config:
camel.component.amqp.includeAmqpAnnotations: true
camel.component.amqp.connectionFactory: org.apache.qpid.jms.JmsConnectionFactory
camel.component.amqp.connectionFactory.remoteURI: amqp://$( kubectl get service eclipse-hono-dispatch-router-ext --output="jsonpath={.spec.clusterIP}" -n hono):15671
camel.component.amqp.username: consumer@HONO
camel.component.amqp.password: verysecret
camel.component.amqp.testConnectionOnStartup: true
camel.source.kafka.topic: mytopic
camel.source.path.destinationType: queue
camel.source.path.destinationName: test-queue
I am getting below error messages:
Could you help me, what values we need to give for the below properties:
camel.component.amqp.connectionFactory: org.apache.qpid.jms.JmsConnectionFactory
camel.component.amqp.connectionFactory.remoteURI: amqp://$( kubectl get service eclipse-hono-dispatch-router-ext --output="jsonpath={.spec.clusterIP}" -n hono):15671
Error message:
2020-12-26 12:50:54,474 INFO Creating task camelamqpsourceconnector-0 (org.apache.kafka.connect.runtime.Worker) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,479 INFO ConnectorConfig values:
config.action.reload = restart
connector.class = org.apache.camel.kafkaconnector.amqp.CamelAmqpSourceConnector
errors.log.enable = false
errors.log.include.messages = false
errors.retry.delay.max.ms = 60000
errors.retry.timeout = 0
errors.tolerance = none
header.converter = null
key.converter = null
name = camelamqpsourceconnector
predicates = []
tasks.max = 1
transforms = []
value.converter = null
(org.apache.kafka.connect.runtime.ConnectorConfig) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,480 INFO EnrichedConnectorConfig values:
config.action.reload = restart
connector.class = org.apache.camel.kafkaconnector.amqp.CamelAmqpSourceConnector
errors.log.enable = false
errors.log.include.messages = false
errors.retry.delay.max.ms = 60000
errors.retry.timeout = 0
errors.tolerance = none
header.converter = null
key.converter = null
name = camelamqpsourceconnector
predicates = []
tasks.max = 1
transforms = []
value.converter = null
(org.apache.kafka.connect.runtime.ConnectorConfig$EnrichedConnectorConfig) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,481 INFO TaskConfig values:
task.class = class org.apache.camel.kafkaconnector.amqp.CamelAmqpSourceTask
(org.apache.kafka.connect.runtime.TaskConfig) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,481 INFO Instantiated task camelamqpsourceconnector-0 with version 0.7.0 of type org.apache.camel.kafkaconnector.amqp.CamelAmqpSourceTask (org.apache.kafka.connect.runtime.Worker) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,482 INFO JsonConverterConfig values:
converter.type = key
decimal.format = BASE64
schemas.cache.size = 1000
schemas.enable = true
(org.apache.kafka.connect.json.JsonConverterConfig) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,483 INFO Set up the key converter class org.apache.kafka.connect.json.JsonConverter for task camelamqpsourceconnector-0 using the worker config (org.apache.kafka.connect.runtime.Worker) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,485 INFO JsonConverterConfig values:
converter.type = value
decimal.format = BASE64
schemas.cache.size = 1000
schemas.enable = true
(org.apache.kafka.connect.json.JsonConverterConfig) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,486 INFO Set up the value converter class org.apache.kafka.connect.json.JsonConverter for task camelamqpsourceconnector-0 using the worker config (org.apache.kafka.connect.runtime.Worker) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,486 INFO Set up the header converter class org.apache.kafka.connect.storage.SimpleHeaderConverter for task camelamqpsourceconnector-0 using the worker config (org.apache.kafka.connect.runtime.Worker) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,492 INFO SourceConnectorConfig values:
config.action.reload = restart
connector.class = org.apache.camel.kafkaconnector.amqp.CamelAmqpSourceConnector
errors.log.enable = false
errors.log.include.messages = false
errors.retry.delay.max.ms = 60000
errors.retry.timeout = 0
errors.tolerance = none
header.converter = null
key.converter = null
name = camelamqpsourceconnector
predicates = []
tasks.max = 1
topic.creation.groups = []
transforms = []
value.converter = null
(org.apache.kafka.connect.runtime.SourceConnectorConfig) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,493 INFO EnrichedConnectorConfig values:
config.action.reload = restart
connector.class = org.apache.camel.kafkaconnector.amqp.CamelAmqpSourceConnector
errors.log.enable = false
errors.log.include.messages = false
errors.retry.delay.max.ms = 60000
errors.retry.timeout = 0
errors.tolerance = none
header.converter = null
key.converter = null
name = camelamqpsourceconnector
predicates = []
tasks.max = 1
topic.creation.groups = []
transforms = []
value.converter = null
(org.apache.kafka.connect.runtime.ConnectorConfig$EnrichedConnectorConfig) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,497 INFO Initializing: org.apache.kafka.connect.runtime.TransformationChain{} (org.apache.kafka.connect.runtime.Worker) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,500 INFO ProducerConfig values:
acks = -1
batch.size = 16384
bootstrap.servers = [my-cluster-kafka-bootstrap:9092]
buffer.memory = 33554432
client.dns.lookup = use_all_dns_ips
client.id = connector-producer-camelamqpsourceconnector-0
compression.type = none
connections.max.idle.ms = 540000
delivery.timeout.ms = 2147483647
enable.idempotence = false
interceptor.classes = []
internal.auto.downgrade.txn.commit = false
key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
linger.ms = 0
max.block.ms = 9223372036854775807
max.in.flight.requests.per.connection = 1
max.request.size = 1048576
metadata.max.age.ms = 300000
metadata.max.idle.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
receive.buffer.bytes = 32768
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
request.timeout.ms = 2147483647
retries = 2147483647
retry.backoff.ms = 100
sasl.client.callback.handler.class = null
sasl.jaas.config = null
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.login.callback.handler.class = null
sasl.login.class = null
sasl.login.refresh.buffer.seconds = 300
sasl.login.refresh.min.period.seconds = 60
sasl.login.refresh.window.factor = 0.8
sasl.login.refresh.window.jitter = 0.05
sasl.mechanism = GSSAPI
security.protocol = PLAINTEXT
security.providers = null
send.buffer.bytes = 131072
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
ssl.endpoint.identification.algorithm = https
ssl.engine.factory.class = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLSv1.3
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
transaction.timeout.ms = 60000
transactional.id = null
value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
(org.apache.kafka.clients.producer.ProducerConfig) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,517 WARN The configuration 'metrics.context.connect.group.id' was supplied but isn't a known config. (org.apache.kafka.clients.producer.ProducerConfig) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,518 WARN The configuration 'metrics.context.connect.kafka.cluster.id' was supplied but isn't a known config. (org.apache.kafka.clients.producer.ProducerConfig) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,518 INFO Kafka version: 2.6.0 (org.apache.kafka.common.utils.AppInfoParser) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,518 INFO Kafka commitId: 62abe01bee039651 (org.apache.kafka.common.utils.AppInfoParser) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,518 INFO Kafka startTimeMs: 1608987054518 (org.apache.kafka.common.utils.AppInfoParser) [StartAndStopExecutor-connect-1-2]
2020-12-26 12:50:54,543 INFO [Worker clientId=connect-1, groupId=connect-cluster-new] Finished starting connectors and tasks (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2020-12-26 12:50:54,600 INFO Starting CamelSourceTask connector task (org.apache.camel.kafkaconnector.CamelSourceTask) [task-thread-camelamqpsourceconnector-0]
2020-12-26 12:50:54,655 INFO [Producer clientId=connector-producer-camelamqpsourceconnector-0] Cluster ID: AxvFw7iiSDCEUx-RwUy0gw (org.apache.kafka.clients.Metadata) [kafka-producer-network-thread | connector-producer-camelamqpsourceconnector-0]
2020-12-26 12:50:54,660 INFO CamelAmqpSourceConnectorConfig values:
camel.aggregation.size = 10
camel.aggregation.timeout = 500
camel.beans.aggregate = null
camel.component.amqp.acceptMessagesWhileStopping = false
camel.component.amqp.acknowledgementModeName = AUTO_ACKNOWLEDGE
camel.component.amqp.allowAutoWiredConnectionFactory = true
camel.component.amqp.allowAutoWiredDestinationResolver = true
camel.component.amqp.allowReplyManagerQuickStop = false
camel.component.amqp.allowSerializedHeaders = false
camel.component.amqp.artemisStreamingEnabled = true
camel.component.amqp.asyncConsumer = false
camel.component.amqp.asyncStartListener = false
camel.component.amqp.asyncStopListener = false
camel.component.amqp.autoStartup = true
camel.component.amqp.autowiredEnabled = true
camel.component.amqp.cacheLevel = null
camel.component.amqp.cacheLevelName = CACHE_AUTO
camel.component.amqp.clientId = null
camel.component.amqp.concurrentConsumers = 1
camel.component.amqp.configuration = null
camel.component.amqp.connectionFactory = org.apache.qpid.jms.JmsConnectionFactory
camel.component.amqp.consumerType = Default
camel.component.amqp.defaultTaskExecutorType = null
camel.component.amqp.destinationResolver = null
camel.component.amqp.disableReplyTo = false
camel.component.amqp.durableSubscriptionName = null
camel.component.amqp.eagerLoadingOfProperties = false
camel.component.amqp.eagerPoisonBody = Poison JMS message due to ${exception.message}
camel.component.amqp.errorHandler = null
camel.component.amqp.errorHandlerLogStackTrace = true
camel.component.amqp.errorHandlerLoggingLevel = WARN
camel.component.amqp.exceptionListener = null
camel.component.amqp.exposeListenerSession = false
camel.component.amqp.headerFilterStrategy = null
camel.component.amqp.idleConsumerLimit = 1
camel.component.amqp.idleTaskExecutionLimit = 1
camel.component.amqp.includeAllJMSXProperties = false
camel.component.amqp.includeAmqpAnnotations = true
camel.component.amqp.jmsKeyFormatStrategy = null
camel.component.amqp.jmsMessageType = null
camel.component.amqp.lazyCreateTransactionManager = true
camel.component.amqp.mapJmsMessage = true
camel.component.amqp.maxConcurrentConsumers = null
camel.component.amqp.maxMessagesPerTask = -1
camel.component.amqp.messageConverter = null
camel.component.amqp.messageCreatedStrategy = null
camel.component.amqp.messageIdEnabled = true
camel.component.amqp.messageListenerContainerFactory = null
camel.component.amqp.messageTimestampEnabled = true
camel.component.amqp.password = verysecret
camel.component.amqp.pubSubNoLocal = false
camel.component.amqp.queueBrowseStrategy = null
camel.component.amqp.receiveTimeout = 1000
camel.component.amqp.recoveryInterval = 5000
camel.component.amqp.replyTo = null
camel.component.amqp.replyToDeliveryPersistent = true
camel.component.amqp.replyToSameDestinationAllowed = false
camel.component.amqp.requestTimeoutCheckerInterval = 1000
camel.component.amqp.selector = null
camel.component.amqp.subscriptionDurable = false
camel.component.amqp.subscriptionName = null
camel.component.amqp.subscriptionShared = false
camel.component.amqp.taskExecutor = null
camel.component.amqp.testConnectionOnStartup = true
camel.component.amqp.transacted = false
camel.component.amqp.transactedInOut = false
camel.component.amqp.transactionManager = null
camel.component.amqp.transactionName = null
camel.component.amqp.transactionTimeout = -1
camel.component.amqp.transferException = false
camel.component.amqp.transferExchange = false
camel.component.amqp.useMessageIDAsCorrelationID = false
camel.component.amqp.username = consumer@HONO
camel.component.amqp.waitForProvisionCorrelationToBeUpdatedCounter = 50
camel.component.amqp.waitForProvisionCorrelationToBeUpdatedThreadSleepingTime = 100
camel.error.handler = default
camel.error.handler.max.redeliveries = 0
camel.error.handler.redelivery.delay = 1000
camel.idempotency.enabled = false
camel.idempotency.expression.header = null
camel.idempotency.expression.type = body
camel.idempotency.kafka.bootstrap.servers = localhost:9092
camel.idempotency.kafka.max.cache.size = 1000
camel.idempotency.kafka.poll.duration.ms = 100
camel.idempotency.kafka.topic = kafka_idempotent_repository
camel.idempotency.memory.dimension = 100
camel.idempotency.repository.type = memory
camel.remove.headers.pattern =
camel.source.camelMessageHeaderKey = null
camel.source.component = amqp
camel.source.contentLogLevel = OFF
camel.source.endpoint.acceptMessagesWhileStopping = false
camel.source.endpoint.acknowledgementModeName = AUTO_ACKNOWLEDGE
camel.source.endpoint.allowReplyManagerQuickStop = false
camel.source.endpoint.allowSerializedHeaders = false
camel.source.endpoint.artemisStreamingEnabled = true
camel.source.endpoint.asyncConsumer = false
camel.source.endpoint.asyncStartListener = false
camel.source.endpoint.asyncStopListener = false
camel.source.endpoint.autoStartup = true
camel.source.endpoint.cacheLevel = null
camel.source.endpoint.cacheLevelName = CACHE_AUTO
camel.source.endpoint.clientId = null
camel.source.endpoint.concurrentConsumers = 1
camel.source.endpoint.connectionFactory = null
camel.source.endpoint.consumerType = Default
camel.source.endpoint.defaultTaskExecutorType = null
camel.source.endpoint.destinationResolver = null
camel.source.endpoint.disableReplyTo = false
camel.source.endpoint.durableSubscriptionName = null
camel.source.endpoint.eagerLoadingOfProperties = false
camel.source.endpoint.eagerPoisonBody = Poison JMS message due to ${exception.message}
camel.source.endpoint.errorHandler = null
camel.source.endpoint.errorHandlerLogStackTrace = true
camel.source.endpoint.errorHandlerLoggingLevel = WARN
camel.source.endpoint.exceptionHandler = null
camel.source.endpoint.exceptionListener = null
camel.source.endpoint.exchangePattern = null
camel.source.endpoint.exposeListenerSession = false
camel.source.endpoint.headerFilterStrategy = null
camel.source.endpoint.idleConsumerLimit = 1
camel.source.endpoint.idleTaskExecutionLimit = 1
camel.source.endpoint.includeAllJMSXProperties = false
camel.source.endpoint.jmsKeyFormatStrategy = null
camel.source.endpoint.jmsMessageType = null
camel.source.endpoint.lazyCreateTransactionManager = true
camel.source.endpoint.mapJmsMessage = true
camel.source.endpoint.maxConcurrentConsumers = null
camel.source.endpoint.maxMessagesPerTask = -1
camel.source.endpoint.messageConverter = null
camel.source.endpoint.messageCreatedStrategy = null
camel.source.endpoint.messageIdEnabled = true
camel.source.endpoint.messageListenerContainerFactory = null
camel.source.endpoint.messageTimestampEnabled = true
camel.source.endpoint.password = null
camel.source.endpoint.pubSubNoLocal = false
camel.source.endpoint.receiveTimeout = 1000
camel.source.endpoint.recoveryInterval = 5000
camel.source.endpoint.replyTo = null
camel.source.endpoint.replyToDeliveryPersistent = true
camel.source.endpoint.replyToSameDestinationAllowed = false
camel.source.endpoint.requestTimeoutCheckerInterval = 1000
camel.source.endpoint.selector = null
camel.source.endpoint.subscriptionDurable = false
camel.source.endpoint.subscriptionName = null
camel.source.endpoint.subscriptionShared = false
camel.source.endpoint.synchronous = false
camel.source.endpoint.taskExecutor = null
camel.source.endpoint.testConnectionOnStartup = false
camel.source.endpoint.transacted = false
camel.source.endpoint.transactedInOut = false
camel.source.endpoint.transactionManager = null
camel.source.endpoint.transactionName = null
camel.source.endpoint.transactionTimeout = -1
camel.source.endpoint.transferException = false
camel.source.endpoint.transferExchange = false
camel.source.endpoint.useMessageIDAsCorrelationID = false
camel.source.endpoint.username = null
camel.source.endpoint.waitForProvisionCorrelationToBeUpdatedCounter = 50
camel.source.endpoint.waitForProvisionCorrelationToBeUpdatedThreadSleepingTime = 100
camel.source.marshal = null
camel.source.maxBatchPollSize = 1000
camel.source.maxPollDuration = 1000
camel.source.path.destinationName = test-queue
camel.source.path.destinationType = queue
camel.source.pollingConsumerBlockTimeout = 0
camel.source.pollingConsumerBlockWhenFull = true
camel.source.pollingConsumerQueueSize = 1000
camel.source.unmarshal = null
camel.source.url = null
topics = test
(org.apache.camel.kafkaconnector.amqp.CamelAmqpSourceConnectorConfig) [task-thread-camelamqpsourceconnector-0]
2020-12-26 12:50:54,905 INFO Setting initial properties in Camel context: [{camel.source.path.destinationName=test-queue, connector.class=org.apache.camel.kafkaconnector.amqp.CamelAmqpSourceConnector, camel.component.amqp.username=consumer@HONO, tasks.max=1, camel.component.amqp.connectionFactory=org.apache.qpid.jms.JmsConnectionFactory, camel.source.component=amqp, camel.component.amqp.testConnectionOnStartup=true, camel.component.amqp.connectionFactory.remoteURI=amqp://$( kubectl get service eclipse-hono-dispatch-router-ext --output="jsonpath={.spec.clusterIP}" -n hono):15671, camel.component.amqp.password=verysecret, camel.component.amqp.includeAmqpAnnotations=true, camel.source.kafka.topic=mytopic, camel.source.path.destinationType=queue, task.class=org.apache.camel.kafkaconnector.amqp.CamelAmqpSourceTask, name=camelamqpsourceconnector}] (org.apache.camel.kafkaconnector.utils.CamelKafkaConnectMain) [task-thread-camelamqpsourceconnector-0]
2020-12-26 12:50:54,985 INFO Using properties from: classpath:application.properties;optional=true (org.apache.camel.main.BaseMainSupport) [task-thread-camelamqpsourceconnector-0]
2020-12-26 12:50:55,175 INFO WorkerSourceTask{id=camelamqpsourceconnector-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask) [task-thread-camelamqpsourceconnector-0]
2020-12-26 12:50:55,175 INFO WorkerSourceTask{id=camelamqpsourceconnector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [task-thread-camelamqpsourceconnector-0]
2020-12-26 12:50:55,176 ERROR WorkerSourceTask{id=camelamqpsourceconnector-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask) [task-thread-camelamqpsourceconnector-0]
org.apache.kafka.connect.errors.ConnectException: Failed to create and start Camel context
at org.apache.camel.kafkaconnector.CamelSourceTask.start(CamelSourceTask.java:144)
at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:232)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:185)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:235)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.IllegalArgumentException: Cannot find getter method: connectionFactory on bean: class org.apache.camel.component.amqp.AMQPComponent when binding property: connectionFactory.remoteURI
at org.apache.camel.support.PropertyBindingSupport.doBuildPropertyOgnlPath(PropertyBindingSupport.java:282)
at org.apache.camel.support.PropertyBindingSupport.doBindProperties(PropertyBindingSupport.java:210)
at org.apache.camel.support.PropertyBindingSupport.access$100(PropertyBindingSupport.java:88)
at org.apache.camel.support.PropertyBindingSupport$Builder.bind(PropertyBindingSupport.java:1785)
at org.apache.camel.main.MainHelper.setPropertiesOnTarget(MainHelper.java:163)
at org.apache.camel.main.BaseMainSupport.autoConfigurationFromProperties(BaseMainSupport.java:1133)
at org.apache.camel.main.BaseMainSupport.autoconfigure(BaseMainSupport.java:424)
at org.apache.camel.main.BaseMainSupport.postProcessCamelContext(BaseMainSupport.java:472)
at org.apache.camel.main.SimpleMain.doInit(SimpleMain.java:32)
at org.apache.camel.support.service.BaseService.init(BaseService.java:83)
at org.apache.camel.support.service.BaseService.start(BaseService.java:111)
at org.apache.camel.kafkaconnector.CamelSourceTask.start(CamelSourceTask.java:141)
... 8 more
2020-12-26 12:50:55,176 ERROR WorkerSourceTask{id=camelamqpsourceconnector-0} Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask) [task-thread-camelamqpsourceconnector-0]
2020-12-26 12:50:55,177 INFO Stopping CamelSourceTask connector task (org.apache.camel.kafkaconnector.CamelSourceTask) [task-thread-camelamqpsourceconnector-0]
2020-12-26 12:50:55,177 INFO CamelSourceTask connector task stopped (org.apache.camel.kafkaconnector.CamelSourceTask) [task-thread-camelamqpsourceconnector-0]
2020-12-26 12:50:55,177 INFO [Producer clientId=connector-producer-camelamqpsourceconnector-0] Closing the Kafka producer with timeoutMillis = 30000 ms. (org.apache.kafka.clients.producer.KafkaProducer) [task-thread-camelamqpsourceconnector-0]