Most messaging patterns covered so far — content-based routers, splitters, message filters — are event-driven. A message arrives, the route processes it, and an output appears almost immediately. But not every data source can push messages to your application. Files land in a directory at unpredictable intervals. Database tables receive new rows from legacy systems that have no event notification mechanism. FTP servers accumulate uploads silently. In all these cases, the only way to detect new data is to ask for it — repeatedly, on a schedule.

The Polling Consumer pattern, described by Hohpe and Woolf, addresses this. A polling consumer actively checks a channel for new messages at regular intervals. It initiates the receive operation — the messaging system does not push messages to it. This is the “pull” model, and it is the natural fit for sources that cannot push: file systems, databases, FTP servers, and scheduled batch jobs.

Apache Camel implements the polling consumer in two ways. Many components — file, ftp, sql, timer — are polling consumers by nature: the from() endpoint polls its source on a configurable schedule. For event-driven routes that occasionally need to pull a message on demand, Camel provides pollEnrich(), which combines a timer trigger with an explicit poll from a second endpoint.

Testing a polling consumer introduces a timing challenge that event-driven patterns do not have. The test cannot simply send a message and expect an immediate response. It must account for the poll interval: the message might sit in the source for seconds before the consumer picks it up. Integration tests with Citrus handle this gracefully — retry-based assertions naturally accommodate the poll delay without resorting to brittle fixed sleeps.

This post covers two variants of the polling consumer: a timer-driven poll from Kafka using pollEnrich(), and a SQL polling consumer that reads rows from PostgreSQL and publishes them to Kafka. The SQL variant is the more realistic and interesting case — it demonstrates database-to-messaging integration with automatic row status updates, and the Citrus test exercises both the database and Kafka sides of the pipeline.

The scenario

Timer-based polling consumer

A timer fires every 10 seconds. On each tick, the route uses pollEnrich() to check a Kafka topic for a waiting message. If a message is available, it is consumed and logged. If no message arrives within a 5-second timeout, the route logs that the poll cycle was empty. This is the simplest form of polling consumer — useful when you want explicit control over when messages are consumed, rather than letting the Kafka client manage the fetch loop.

SQL polling consumer

A more common real-world scenario: a database table receives new order rows from a legacy system. There is no event notification — the Camel route polls the orders.orders table every 30 seconds, looking for rows with status = 'PLACED'. Each row is read, marshalled to JSON, and published to a Kafka topic. After successful processing, the onConsume query updates the row’s status to 'PROCESSING', preventing it from being picked up again on the next poll.

This two-phase behavior — read then update — is exactly what makes the SQL polling consumer worth testing end-to-end. A unit test could verify the SQL query syntax, but only an integration test proves that the row is actually read, the Kafka message is published, and the status update takes effect.

The Camel routes

Timer-based polling consumer

Quarkus

@ApplicationScoped
public class PollingConsumerRoute extends RouteBuilder {

    @Override
    public void configure() {
        from("timer:poll-trigger?period=10000&delay=5000")
            .routeId("polling-consumer")
            .log("Polling consumer triggered — checking for messages …")
            .pollEnrich("kafka:eip.consumer.poll?brokers={{kafka.brokers}}"
                + "&groupId=polling-consumer&autoOffsetReset=earliest", 5000)
            .choice()
                .when(body().isNull())
                    .log("No message available during this poll cycle")
                .otherwise()
                    .unmarshal().json()
                    .log("Polled message: order ${body[order_id]}, type=${body[event_type]}")
            .end();
    }
}

The route starts from a timer, not from Kafka directly. Every 10 seconds, the timer fires and pollEnrich() attempts to pull one message from the eip.consumer.poll topic with a 5-second timeout. The choice() block handles the two outcomes: a null body means no message was available; otherwise, the message is deserialized and logged.

This is fundamentally different from a standard from("kafka:...") consumer. A Kafka from() consumer is event-driven — Camel manages the poll loop internally and your route logic runs whenever messages arrive. With pollEnrich(), you control the timing: the poll happens exactly when the timer fires, and at most one message is consumed per cycle.

Spring Boot

@Component
public class PollingConsumerRoute extends RouteBuilder {

    @Override
    public void configure() {
        from("timer:poll-trigger?period=10000&delay=5000")
            .routeId("polling-consumer")
            .log("Polling consumer triggered — checking for messages …")
            .pollEnrich("kafka:eip.consumer.poll?brokers={{kafka.brokers}}"
                + "&groupId=polling-consumer&autoOffsetReset=earliest", 5000)
            .choice()
                .when(body().isNull())
                    .log("No message available during this poll cycle")
                .otherwise()
                    .unmarshal().json()
                    .log("Polled message: order ${body[order_id]}, type=${body[event_type]}")
            .end();
    }
}

@Component replaces @ApplicationScoped — the route logic is identical.

SQL polling consumer

The SQL polling consumer is more interesting because it bridges two infrastructure systems: PostgreSQL as the source and Kafka as the destination.

Quarkus

@ApplicationScoped
public class SqlPollingConsumerRoute extends RouteBuilder {

    @Override
    public void configure() {
        from("sql:SELECT * FROM orders.orders WHERE status = 'PLACED' "
                + "ORDER BY created_at LIMIT 10"
                + "?delay=30000"
                + "&onConsume=UPDATE orders.orders SET status = 'PROCESSING' WHERE id = :#id")
            .routeId("polling-consumer-sql")
            .log("SQL Polling Consumer — processing order from DB: "
                + "id=${body[id]}, customer=${body[customer_id]}, sku=${body[item_sku]}")
            .marshal().json()
            .to("kafka:eip.orders.placed?brokers={{kafka.brokers}}");
    }
}

The from("sql:...") component is a natural polling consumer. Every 30 seconds (controlled by delay=30000), Camel executes the SELECT query and creates one exchange per row. The query selects at most 10 rows ordered by creation time — this is the backpressure mechanism, preventing the consumer from overwhelming the downstream Kafka topic during a burst of new orders.

The onConsume parameter is the key to reliable processing. After each row is successfully processed (marshalled to JSON and published to Kafka), Camel executes the UPDATE statement, setting the row’s status from 'PLACED' to 'PROCESSING'. The :#id syntax is a named parameter that Camel resolves from the current exchange body — it refers to the id column of the row being processed. On the next poll cycle, the WHERE clause status = 'PLACED' naturally excludes already-processed rows.

Spring Boot

@Component
public class SqlPollingConsumerRoute extends RouteBuilder {

    @Override
    public void configure() {
        from("sql:SELECT * FROM orders.orders WHERE status = 'PLACED' "
                + "ORDER BY created_at LIMIT 10"
                + "?delay=30000"
                + "&onConsume=UPDATE orders.orders SET status = 'PROCESSING' WHERE id = :#id")
            .routeId("polling-consumer-sql")
            .log("SQL Polling Consumer — processing order from DB: "
                + "id=${body[id]}, customer=${body[customer_id]}, sku=${body[item_sku]}")
            .marshal().json()
            .to("kafka:eip.orders.placed?brokers={{kafka.brokers}}");
    }
}

YAML DSL

- route:
    id: polling-consumer-sql
    from:
      uri: "sql:SELECT * FROM orders.orders WHERE status = 'PLACED' ORDER BY created_at LIMIT 10"
      parameters:
        delay: 30000
        onConsume: "UPDATE orders.orders SET status = 'PROCESSING' WHERE id = :#id"
      steps:
        - log: "SQL Polling Consumer — processing order from DB: id=${body[id]}, customer=${body[customer_id]}, sku=${body[item_sku]}"
        - marshal:
            json:
              library: Jackson
        - to:
            uri: "kafka:eip.orders.placed"
            parameters:
              brokers: "{{kafka.brokers}}"

Polling parameters

Camel’s polling consumers share a common set of scheduling parameters:

Parameter Description Default
delay Milliseconds between polls 500
maxMessagesPerPoll Maximum messages to process per poll cycle 0 (unlimited)
greedy Poll again immediately if messages were found false
sendEmptyMessageWhenIdle Send an empty exchange if no messages found false

The greedy option is particularly useful for database polling. With greedy=true and a high delay, the consumer polls rapidly when rows are available but backs off to the long interval when the table is empty. This avoids the wasted CPU and database connections that come from aggressive short-interval polling on an empty table.

The polling consumer test

The timer-based polling consumer test is straightforward: send a message to Kafka, then wait for the poll cycle to pick it up.

Quarkus test

@QuarkusTest
@CitrusSupport
class EipTests implements EipTestSupport {

    @CitrusResource
    TestCaseRunner t;

    @Inject
    @BindToRegistry
    CamelContext camelContext;

    @Nested
    class PollingConsumerTest {

        @Test
        public void shouldPollMessageFromKafka() {
            t.given(
                createVariables()
                    .variable("id", "citrus:randomNumber(4)")
                    .variable("eventType", "order_placed")
                    .variable("amount", 99)
            );

            t.given(waitForCamelRouteStarted("polling-consumer", camelContext));

            t.when(
                send()
                    .endpoint("kafka:eip.consumer.poll")
                    .message()
                    .body(Resources.create("templates/order.json"))
                    .header(KafkaMessageHeaders.MESSAGE_KEY, "org.citrusframework:citrus-website:pom:1.1.0")
            );

            t.then(verifyRouteStats("polling-consumer", """
                    { "exchangesCompleted": "@greaterThan(1)@" }
                """, camelContext));
        }
    }
}

Given — set up variables and wait for the route. The test generates a random order ID and sets up the event type and amount for the order template. It then waits for the polling-consumer route to reach Started status — important because the timer-driven route needs to be active and the pollEnrich() Kafka consumer group needs to be registered before a message can be consumed.

When — send an order to the polled topic. A single order message goes to kafka:eip.consumer.poll — the topic that the pollEnrich() call reads from.

Then — assert the route processed exchanges. The test does not try to receive the message from another Kafka topic, because the polling consumer route only logs the message — it does not forward it anywhere. Instead, it uses verifyRouteStats with a JSON stats expression { "exchangesCompleted": "@greaterThan(1)@" }. The @greaterThan(1)@ matcher checks that more than one exchange has completed on the polling-consumer route. Why more than one? Because the timer fires repeatedly, producing exchanges even when no message is available (the “no message” branch of the choice()). The assertion retries until the route has processed at least one exchange with the polled message.

This is a key insight for testing polling consumers: when the consumer only logs or internally processes the message without producing output on a verifiable endpoint, Camel’s management API provides an alternative verification path.

The SQL polling consumer test

The SQL polling consumer test is richer than the timer-based variant because it exercises two infrastructure systems and verifies three things: a database row is consumed, a Kafka message is produced, and the database row is updated.

Quarkus test

@QuarkusTest
@CitrusSupport
class EipTests implements EipTestSupport {

    @CitrusResource
    TestCaseRunner t;

    @Inject
    @BindToRegistry
    DataSource dataSource;

    @Inject
    @BindToRegistry
    CamelContext camelContext;

    @Nested
    class SqlPollingConsumerTest {

        @Test
        public void shouldHandleSqlPollingConsumer() {
            t.given(
                createVariables()
                    .variable("id", "citrus:randomNumber(4)")
                    .variable("amount", 40)
            );

            t.given(waitForCamelRouteStarted("polling-consumer-sql", camelContext));

            t.when(
                sql(dataSource)
                    .statement("INSERT INTO orders.orders "
                        + "(customer_id, item_sku, quantity, amount) "
                        + "VALUES ('CUST-00org.citrusframework:citrus-website:pom:1.1.0', 'SKU-org.citrusframework:citrus-website:pom:1.1.0', '1', '${amount}')")
            );

            t.then(
                repeatOnError()
                    .times(15)
                    .actions(
                        receive()
                            .endpoint("kafka:eip.orders.placed?consumerGroup=citrus-placed-group")
                            .message()
                            .body("""
                            {
                              "id": "@variable(order_id)@",
                              "customer_id": "CUST-00org.citrusframework:citrus-website:pom:1.1.0",
                              "status": "PLACED",
                              "amount": ${amount}.0,
                              "item_sku": "SKU-org.citrusframework:citrus-website:pom:1.1.0",
                              "quantity": 1,
                              "created_at": "@ignore@"
                            }
                            """)
                    )
            );

            t.then(
                sql(dataSource)
                    .query()
                    .statement("SELECT status FROM orders.orders WHERE id = '${order_id}'")
                    .validate("status", "PROCESSING")
            );
        }
    }
}

This test has four distinct phases that exercise the full polling consumer pipeline.

Given — set up variables and wait for the route. The test generates a random ID used to construct a unique customer_id and item_sku. It then waits for the polling-consumer-sql route to start. This is critical: if the INSERT happens before the route is polling, the row could sit in the database for an entire 30-second delay interval before being picked up.

When — insert a test row into PostgreSQL. Citrus’s sql() action inserts a new order row directly into the orders.orders table. The row is created with the default status 'PLACED', making it immediately eligible for the SQL consumer’s SELECT query. This is the test’s “stimulus” — instead of sending a message to a Kafka topic, the test writes directly to the database that the polling consumer reads from.

Then (first assertion) — verify the Kafka output. The test receives a message from kafka:eip.orders.placed — the topic that the SQL polling consumer publishes to. The repeatOnError() block retries up to 15 times with 1-second intervals, giving the poll cycle time to fire and process the row.

The expected body is an inline JSON template that validates the row’s data. Two Citrus features are at work here:

  • @variable(order_id)@ captures the database-generated id value into a Citrus variable named order_id rather than asserting a specific value. The ID is auto-generated by PostgreSQL’s SERIAL column, so the test cannot predict it in advance. By extracting it into a variable, the test can reference it in subsequent assertions.
  • @ignore@ skips validation of the created_at timestamp, which depends on the database server’s clock.

Then (second assertion) — verify the database status update. The final sql().query() action reads the row back from PostgreSQL using the captured ${order_id} and asserts that its status has changed from 'PLACED' to 'PROCESSING'. This verifies the onConsume query — the UPDATE that Camel executes after successfully processing each row.

This second assertion is what makes the test truly end-to-end. Without it, you would know that the row was read and a Kafka message was produced, but you would not know whether the status update actually ran. A failure in the onConsume query would cause the same row to be processed again on the next poll cycle — a subtle bug that only an integration test covering both the read and the update can catch.

For the test infrastructure setup, shared test utilities, runtime wiring, dependencies, and how to run the tests, see the Camel EIP examples overview page.

Why polling consumer tests are different

Polling consumer tests have characteristics that set them apart from event-driven pattern tests:

The poll interval is the test’s timing constraint. With an event-driven consumer, a message sent to Kafka is processed within milliseconds. With a polling consumer, the message sits in the source — a Kafka topic, a database table, a file directory — until the next poll cycle fires. The test must wait at least one full interval. Citrus’s repeatOnError() and timeout parameters handle this gracefully, retrying until the poll cycle picks up the test data.

The test produces input in a different medium than the route consumes. For the SQL polling consumer, the test does not send a Kafka message — it inserts a database row. The test and the route operate on different infrastructure: the test writes to PostgreSQL, the route reads from PostgreSQL and writes to Kafka, and the test verifies on Kafka. This cross-infrastructure flow is a natural fit for integration testing and would be impossible to verify with unit tests alone.

Side effects are first-class assertions. The onConsume status update is a side effect that is just as important as the Kafka output. The test verifies it explicitly with a SQL query after receiving the Kafka message. If the onConsume query fails silently, the row would be reprocessed on the next poll — a production bug that only a test covering both the output and the side effect can detect.

verifyRouteStats provides a fallback verification path. When a polling consumer only logs messages internally (like the timer-based variant), there is no output endpoint to receive from. Citrus’s route statistics verification lets the test verify that the route processed exchanges without errors — a useful technique for any route that does not produce verifiable output on an external endpoint.

Key takeaways

  • Polling consumers are pull-based. They actively check for new data on a schedule, making them the right choice for sources that cannot push: databases, files, FTP, and scheduled batch jobs. The poll interval and maxMessagesPerPoll provide natural backpressure control.
  • pollEnrich() turns any route into a polling consumer. A timer-triggered route with pollEnrich() gives you explicit control over when messages are consumed — useful when you need to decouple the consumption schedule from the source’s availability.
  • SQL polling with onConsume is a two-phase operation. The SELECT reads the row; the onConsume UPDATE marks it as processed. Both phases must be verified in the test — the Kafka output proves the read worked, and the SQL assertion proves the update ran.
  • @variable(name)@ captures dynamic values from the system under test. Database-generated IDs cannot be predicted by the test. Citrus’s variable extraction lets you capture these values from the first assertion and use them in subsequent verifications.
  • repeatOnError() accommodates poll timing naturally. Rather than inserting fixed sleeps matching the poll interval, Citrus retries the assertion until it succeeds or the retry limit is reached — making tests both reliable and as fast as possible.
  • verifyRouteStats verifies routes without external output. When a polling consumer only logs or internally processes messages, Citrus’s Camel route statistics verification provides exchange counts as an alternative verification mechanism, with support for flexible matchers like @greaterThan()@.
  • Three runtimes, one test pattern. Whether you run on Quarkus, Spring Boot, or YAML DSL with Camel JBang, the test structure — seed the source, wait for the poll, verify the output and side effects — stays the same. Only the bootstrap annotations and infrastructure wiring change.