SQL

The component level is the highest level which holds general and common configurations that are inherited by the endpoints. For example a component may have security settings, credentials for authentication, urls for network connection and so forth.

Some components only have a few options, and others may have many. Because components typically have pre configured defaults that are commonly used, then you may often only need to configure a few options on a component; or none at all.

Configuring components can be done with the Component DSL, in a configuration file (application.properties|yaml), or directly with Java code.

Where you find yourself configuring the most is on endpoints, as endpoints often have many options, which allows you to configure what you need the endpoint to do. The options are also categorized into whether the endpoint is used as consumer (from) or as a producer (to), or used for both.

Configuring endpoints is most often done directly in the endpoint URI as path and query parameters. You can also use the Endpoint DSL and DataFormat DSL as a type safe way of configuring endpoints and data formats in Java.

A good practice when configuring options is to use Property Placeholders, which allows to not hardcode urls, port numbers, sensitive information, and other settings. In other words placeholders allows to externalize the configuration from your code, and gives more flexibility and reuse.

The following two sections lists all the options, firstly for the component followed by the endpoint.

When using the SQL component as consumer, you can now also use expression parameters (simple language) to build dynamic query parameters, such as calling a method on a bean to retrieve an id, date or something.

For example in the sample below we call the nextId method on the bean myIdGenerator:

And the bean has the following method:

Notice that there is no existing Exchange with message body and headers, so the simple expression you can use in the consumer are most useable for calling bean methods as in this example.

You have a few options to tune the org.apache.camel.processor.idempotent.jdbc.JdbcMessageIdRepository for your needs:

The option tableName can be used to use the default SQL queries but with a different table name. However if you want to customize the SQL queries then you can configure each of them individually.

One of the limitations of org.apache.camel.processor.idempotent.jdbc.JdbcMessageIdRepository is that it does not handle orphan locks resulting from JVM crash or non graceful shutdown. This can result in unprocessed files/messages if this is implementation is used with camel-file, camel-ftp etc. if you need to address orphan locks processing then use org.apache.camel.processor.idempotent.jdbc.JdbcOrphanLockAwareIdempotentRepository. This repository keeps track of the locks held by an instance of the application. For each lock held, the application will send keep alive signals to the lock repository resulting in updating the createdAt column with the current Timestamp. When an application instance tries to acquire a lock if the, then there are three possibilities exist :

This repository has two additional configuration parameters

Some SQL implementations are not fast on a per query basis. The JdbcMessageIdRepository implementation does its idempotent checks individually within SQL transactions. Checking a mere 100 keys can take minutes. The JdbcCachedMessageIdRepository preloads an in-memory cache on start with the entire list of keys. This cache is then checked first before passing through to the original implementation.

As with all cache implementations, there are considerations that should be made with regard to stale data and your specific usage.

To be operational, each aggregator uses two table: the aggregation and completed one. By convention the completed has the same name as the aggregation one suffixed with "_COMPLETED". The name must be configured in the Spring bean with the RepositoryName property. In the following example aggregation will be used.

The table structure definition of both table are identical: in both case a String value is used as key (id) whereas a Blob contains the exchange serialized in byte array.
However one difference should be remembered: the id field does not have the same content depending on the table.
In the aggregation table id holds the correlation Id used by the component to aggregate the messages. In the completed table, id holds the id of the exchange stored in corresponding the blob field.

Here is the SQL query used to create the tables, just replace "aggregation" with your aggregator repository name.

Since they can contain any type of payload, Exchanges are not serializable by design. It is converted into a byte array to be stored in a database BLOB field. All those conversions are handled by the JdbcCodec class. One detail of the code requires your attention: the ClassLoadingAwareObjectInputStream.

The ClassLoadingAwareObjectInputStream has been reused from the Apache ActiveMQ project. It wraps an ObjectInputStream and use it with the ContextClassLoader rather than the currentThread one. The benefit is to be able to load classes exposed by other bundles. This allows the exchange body and headers to have custom types object references.

While deserializing it’s important to notice that the decode function and the unmarshallExchange method will allow only all java packages and subpackages and org.apache.camel packages and subpackages. The remaining classes will be blacklisted. So you’ll need to change the filter in case of need. This could be accomplished by changing the deserializationFilter field on the repository.

A Spring PlatformTransactionManager is required to orchestrate transaction.

Depending on the targeted environment, the aggregator might need some configuration. As you already know, each aggregator should have its own repository (with the corresponding pair of table created in the database) and a data source. If the default lobHandler is not adapted to your database system, it can be injected with the lobHandler property.

Here is the declaration for Oracle:

You can turn on optimisticLocking and use this JDBC based aggregation repository in a clustered environment where multiple Camel applications shared the same database for the aggregation repository. If there is a race condition there JDBC driver will throw a vendor specific exception which the JdbcAggregationRepository can react upon. To know which caused exceptions from the JDBC driver is regarded as an optimistic locking error we need a mapper to do this. Therefore there is a org.apache.camel.processor.aggregate.jdbc.JdbcOptimisticLockingExceptionMapper allows you to implement your custom logic if needed. There is a default implementation org.apache.camel.processor.aggregate.jdbc.DefaultJdbcOptimisticLockingExceptionMapper which works as follows:

The following check is done:

You can in addition add FQN classnames, and if any of the caused exception (or any nested) equals any of the FQN class names, then its an optimistic locking error.

Here is an example, where we define 2 extra FQN class names from the JDBC vendor.

JdbcAggregationRepository uses two distinct transaction templates from Spring-TX. One is read-only and one is used for read-write operations.

However, when using JdbcAggregationRepository within a route that itself uses <transacted /> and there’s common PlatformTransactionManager used, there may be a need to configure propagation behavior used by transaction templates inside JdbcAggregationRepository.

Here’s a way to do it:

Propagation is specified by constants of org.springframework.transaction.TransactionDefinition interface, so propagationBehaviorName is convenient setter that allows to use names of the constants.

JdbcAggregationRepository does not provide recovery in a clustered environment.

You may use ClusteredJdbcAggregationRepository that provides a limited support for recovery in a clustered environment : recovery mechanism is dealt separately by members of the cluster, i.e. a member may only recover exchanges that it completed itself.

To enable this behaviour, property recoverByInstance must be set to true, and instanceId property must be defined using a unique identifier (a string) for each member of the cluster.

Besides, completed table must have a instance_id VARCHAR(255) column.

N.B. : since each member is the only responsible for the recovery of its completed exchanges, if a member is stopped, its completed exchanges will not be recovered until it is restarted, unless you update completed table to affect them to another member (by changing instance_id for those completed exchanges).

There’s special database that may cause problems with optimistic locking used by JdbcAggregationRepository. PostgreSQL marks connection as invalid in case of data integrity violation exception (the one with SQLState 23505). This makes the connection effectively unusable within nested transaction. Details can be found in this document.

org.apache.camel.processor.aggregate.jdbc.PostgresAggregationRepository extends JdbcAggregationRepository and uses special INSERT .. ON CONFLICT .. statement to provide optimistic locking behavior.

This statement is (with default aggregation table definition):

Details can be found in PostgreSQL documentation.

When this clause is used, java.sql.PreparedStatement.executeUpdate() call returns 0 instead of throwing SQLException with SQLState=23505. Further handling is exactly the same as with generic JdbcAggregationRepository, but without marking PostgreSQL connection as invalid.