Add JDBC examples

[oscerd]

to camel-kafka-connector-examples repo

[orpiske]

Providing these, as taken from the test code for a kick-off. Maybe we can leverage it ...

Base configuration using Postgres, but should work in the same way for other DBs:

connector.class=org.apache.camel.kafkaconnector.jdbc.CamelJdbcSinkConnector
tasks.max=1
camel.sink.path.dataSourceName=someName
topics=org.apache.camel.kafkaconnector.jdbc.sink.CamelSinkJDBCITCase
name=CamelJDBCSinkConnector
value.converter=org.apache.kafka.connect.storage.StringConverter
camel.component.jdbc.dataSource=#class:org.apache.camel.kafkaconnector.jdbc.services.TestDataSource
key.converter=org.apache.kafka.connect.storage.StringConverter
camel.sink.endpoint.useHeadersAsParameters=true

You need to provide a data source. This should setup the JDBC url, username and password. Here's an example from the test code:

public class TestDataSource extends PGSimpleDataSource {
    private static final Logger LOG = LoggerFactory.getLogger(TestDataSource.class);

    private static final String URL;

    static {
        URL = System.getProperty(JDBCProperties.JDBC_CONNECTION_URL);
    }

    public TestDataSource() {
        super();
        setUrl(URL);

        setUser("ckc");
        setPassword("ckcDevel123");

    }
}

The database insertion is handled by the body of the message and the rows will be resolved using Camel headers, such as:

String body = "insert into test(test_name, test_data) values(:?TestName, :?TestData)"; 

... 


Map<String, String> jdbcParameters = new HashMap<>();

// The prefix 'CamelHeader' is removed by the SinkTask
jdbcParameters.put(CamelSinkTask.HEADER_CAMEL_PREFIX + "TestName", "SomeName1");
jdbcParameters.put(CamelSinkTask.HEADER_CAMEL_PREFIX + "TestData", "test data ");


// Send the data
kafkaClient.produce("topic-name", body, jdbcParameters);

[catchkuo]

Implement the TestDataSource is necessary?
Can I set the JDBC_CONNECTION_URL include DB account,password in the properties file? Like normal JDBC or other KAFKA connectors. Because that's mean have to compile again when I want change the DB config.

[oscerd]

I believe you can do like we do in camel-sql connector

https://github.com/apache/camel-kafka-connector-examples/blob/master/sql/sql-sink/config/CamelSqlSinkConnector.properties

[catchkuo]

Thanks. I will try it.

[oscerd]

Here is a little example:
https://github.com/apache/camel-kafka-connector-examples/tree/master/jdbc/jdbc-sink

[oscerd]

Done.