Exception on endpoint being treated as a polling consumer before createPollingConsumer is called.

[mgenereu]

I'm getting an exception on this line in the code. Best I can tell, it's trying to get the endpoint from the Camel Context before the endpoint has been upgraded to a PollingConsumer on the line afterwards. direct doesn't have these parameters.

camel-kafka-connector/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java

Line 109 in 4a73d04

Endpoint endpoint = cms.getEndpoint(localUrl);

org.apache.camel.ResolveEndpointFailedException: Failed to resolve endpoint: direct://end?pollingConsumerBlockTimeout=0&pollingConsumerBlockWhenFull=true&pollingConsumerQueueSize=1000 due to: Error binding property (pollingConsumerBlockTimeout=0) with name: pollingConsumerBlockTimeout on bean: direct://end?pollingConsumerBlockTimeout=0&pollingConsumerBlockWhenFull=true&pollingConsumerQueueSize=1000 with value: 0

[oscerd]

What is the configuration of the connector? What connector are you using?

[mgenereu]

I built a JAR with this:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>kafka-test</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.apache.camel.kafkaconnector</groupId>
            <artifactId>camel-salesforce-kafka-connector</artifactId>
            <version>0.4.0</version>
        </dependency>
    </dependencies>
    <build>
        <finalName>salesforce-connector</finalName>
        <plugins>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.6.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.1.1</version>

                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>

                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>

            </plugin>
        </plugins>
    </build>
</project>

Then I create a container fdd with this (fat jar is in connectors):

FROM lensesio/fast-data-dev:latest
COPY connectors /connectors

Then I run the container:

docker run --rm -p 2181:2181 -p 3030:3030 -p 8081-8083:8081-8083 -p 9581-9585:9581-9585 -p 9092:9092 -e CONNECT_PLUGIN_PATH=/connectors fdd

Then I add the connector to Kafka Connect via the UI:

name=CamelSalesforceSourceConnector
connector.class=org.apache.camel.kafkaconnector.salesforce.CamelSalesforceSourceConnector
tasks.max=1

[mgenereu]

Here's the whole error message as well:

org.apache.kafka.connect.errors.ConnectException: Failed to create and start Camel context
	at org.apache.camel.kafkaconnector.CamelSourceTask.start(CamelSourceTask.java:103)
	at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:202)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.camel.ResolveEndpointFailedException: Failed to resolve endpoint: direct://end?pollingConsumerBlockTimeout=0&pollingConsumerBlockWhenFull=true&pollingConsumerQueueSize=1000 due to: Error binding property (pollingConsumerBlockTimeout=0) with name: pollingConsumerBlockTimeout on bean: direct://end?pollingConsumerBlockTimeout=0&pollingConsumerBlockWhenFull=true&pollingConsumerQueueSize=1000 with value: 0
	at org.apache.camel.impl.engine.AbstractCamelContext.doGetEndpoint(AbstractCamelContext.java:876)
	at org.apache.camel.impl.engine.AbstractCamelContext.getEndpoint(AbstractCamelContext.java:766)
	at org.apache.camel.kafkaconnector.utils.CamelMainSupport.getEndpoint(CamelMainSupport.java:142)
	at org.apache.camel.kafkaconnector.CamelSourceTask.start(CamelSourceTask.java:96)
	... 8 more
Caused by: org.apache.camel.PropertyBindingException: Error binding property (pollingConsumerBlockTimeout=0) with name: pollingConsumerBlockTimeout on bean: direct://end?pollingConsumerBlockTimeout=0&pollingConsumerBlockWhenFull=true&pollingConsumerQueueSize=1000 with value: 0
	at org.apache.camel.support.PropertyBindingSupport.bindProperty(PropertyBindingSupport.java:630)
	at org.apache.camel.support.PropertyBindingSupport.lambda$doBindProperties$2(PropertyBindingSupport.java:608)
	at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
	at java.util.stream.SortedOps$SizedRefSortingSink.end(SortedOps.java:357)
	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:483)
	at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
	at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
	at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
	at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:485)
	at org.apache.camel.support.PropertyBindingSupport.doBindProperties(PropertyBindingSupport.java:573)
	at org.apache.camel.support.PropertyBindingSupport.access$000(PropertyBindingSupport.java:71)
	at org.apache.camel.support.PropertyBindingSupport$Builder.bind(PropertyBindingSupport.java:258)
	at org.apache.camel.support.DefaultEndpoint.setProperties(DefaultEndpoint.java:433)
	at org.apache.camel.support.DefaultEndpoint.configureProperties(DefaultEndpoint.java:401)
	at org.apache.camel.support.DefaultComponent.setProperties(DefaultComponent.java:453)
	at org.apache.camel.component.direct.DirectComponent.createEndpoint(DirectComponent.java:51)
	at org.apache.camel.support.DefaultComponent.createEndpoint(DefaultComponent.java:237)
	at org.apache.camel.impl.engine.AbstractCamelContext.doGetEndpoint(AbstractCamelContext.java:844)
	... 11 more
Caused by: java.lang.IllegalArgumentException: Could not find a suitable setter for property: pollingConsumerBlockTimeout as there isn't a setter method with same type: java.lang.String nor type conversion possible: No type converter available to convert from type: java.lang.String to the required type: long with value 0
	at org.apache.camel.support.IntrospectionSupport.setProperty(IntrospectionSupport.java:779)
	at org.apache.camel.impl.engine.DefaultBeanIntrospection.setProperty(DefaultBeanIntrospection.java:195)
	at org.apache.camel.support.PropertyBindingSupport.setProperty(PropertyBindingSupport.java:734)
	at org.apache.camel.support.PropertyBindingSupport.bindProperty(PropertyBindingSupport.java:627)
	... 29 more

Thanks for taking a glance!

[valdar]

@mgenereu you can have a look at how things are handled here: https://github.com/apache/camel-kafka-connector/tree/master/tests/itests-salesforce in the salesforce integration test, might be @orpiske can give you more insight in what has been done there.

[mgenereu]

@mgenereu you can have a look at how things are handled here: https://github.com/apache/camel-kafka-connector/tree/master/tests/itests-salesforce in the salesforce integration test, might be @orpiske can give you more insight in what has been done there.

Thank you. I'll take another look at that project. In the meantime I've switched over to regular Camel Main and everything is working fine. Obviously my above example is missing some configuration, but it shouldn't fail where it is failing.

[orpiske]

@mgenereu feel free to get in touch if you need any insight about those tests. Also, if you could provide both the whole log as well as the contents of the connector (ie.: what jars are bundled with the package). I'd love to take a look at it and see if there's something we could add to our tests to cover the scenario are trying on.

[mgenereu]

I think it has to more about how I'm creating the project or jar than any specific connector. I get the same error with the camel-timer-kafka-connector instead of the camel-salesforce-kafka-connector.

[oscerd]

@mgenereu can you share a reproduce on github with all the stuff you're doing? So we could try to dig a bit.

[mgenereu]

Will do! Thanks!

[mgenereu]

Created: https://github.com/mgenereu/apache-camel-kafka-source-connector-test
Referenced @hguerrero as he might be interested as well.

[mgenereu]

@andmarios might also be interested how this unfolds since https://github.com/lensesio/fast-data-dev may be different than https://github.com/strimzi/strimzi-kafka-operator .

[mgenereu]

@mgenereu can you share a reproduce on github with all the stuff you're doing? So we could try to dig a bit.

@oscerd, was the repository to your satisfaction?

[oscerd]

Yes, thanks. I'll try to have a look, but I won't be around for a while. Maybe @valdar and @orpiske could help. Or I'll take a look when I'll come back

[mgenereu]

No rush. Just wanted to set you up for success. Thank you!

[orpiske]

Just to ACK that I am also keeping an eye on this one and will try to take a look at it at the first opportunity.