KafkaIdempotentRepositoryEagerIT.java

Overview

`KafkaIdempotentRepositoryEagerIT` is an integration test class designed to verify the behavior of the **KafkaIdempotentRepository** when used in an "eager" idempotent consumer scenario within Apache Camel routes. The class ensures that duplicate messages are properly filtered out by the idempotent repository backed by a Kafka topic, and tests how the repository behaves when exceptions occur during message processing.

This test extends `SimpleIdempotentTest` (presumably a base class for idempotent consumer tests) and leverages Camel's testing framework features such as routes, mock endpoints, and producer templates.

Class Details

KafkaIdempotentRepositoryEagerIT

• Package: org.apache.camel.processor.idempotent.kafka

• Extends: SimpleIdempotentTest

• Purpose: To perform integration tests confirming the correct behavior of Kafka-based idempotent repository with eager evaluation.

• Key Features Tested:

  • Duplicate message filtering by idempotent consumer.

  • Rollback behavior on processing exceptions.

Fields

Methods

createRepositoryTopic()

• Type: static void

• Annotations: @BeforeAll

• Description:
  Creates the Kafka topic used by the idempotent repository before any tests run. Uses KafkaTestUtil to ensure the topic exists with one partition.

• Usage:
  Automatically executed once before all tests.

createRouteBuilder()

• Type: RouteBuilder

• Overrides: SimpleIdempotentTest#createRouteBuilder()

• Description:
  Defines a Camel route used during testing:

  • Reads from endpoint "direct:in".

  • Sends to "mock:before" endpoint (for message count before idempotent filtering).

  • Applies idempotentConsumer based on the message header "id".

  • Uses the kafkaIdempotentRepositoryEager bean for idempotent store.

  • Sends unique messages to "mock:out" endpoint.

• Returns:
  A configured RouteBuilder instance.

• Usage:
  Used internally by Camel test framework to set up the route.

testRemovesDuplicates()

• Type: void

• Annotations: @Test

• Description:
  Sends 10 messages with header "id" cycling through values 0 to 4.
  Verifies that:

  • The "mock:before" endpoint receives all 10 messages (before filtering).

  • The "mock:out" endpoint receives only 5 messages (duplicates removed).

• Assertions:

  • mockOut.getReceivedCounter() == 5

  • mockBefore.getReceivedCounter() == 10

• Usage Example:

testRemovesDuplicates();

testRollsBackOnException()

• Type: void

• Annotations: @Test

• Description:
  Tests behavior when message processing throws an exception:

  • Configures "mock:out" to throw an IllegalArgumentException when header "id" is 0.

  • Sends 10 messages with "id" cycling 0 to 4.

  • Expects exceptions on certain messages and verifies rollback behavior.

• Assertions:

  • "mock:out" still has exactly 5 messages processed successfully.

  • "mock:before" endpoint has seen all 20 messages (10 from previous test + 10 here).

• Usage Example:

testRollsBackOnException();

Important Implementation Details

• Idempotent Repository Setup:
  The repository topic name is unique per test class instance using a UUID suffix to avoid interference between concurrent tests.

• Eager Idempotent Consumer:
  The route uses the idempotentConsumer pattern with eager evaluation, meaning the idempotent check happens before processing downstream. This ensures duplicates are filtered as early as possible.

• Exception Handling:
  When an exception occurs during message processing, the idempotent repository should rollback and not mark messages as processed, allowing retries.

• Kafka as Backend:
  The idempotent repository stores processed message IDs in a Kafka topic. This supports distributed, scalable idempotency tracking.

Interaction with Other Components

• KafkaIdempotentRepository:
  This class under test is a specialized idempotent repository that stores keys in a Kafka topic.

• KafkaTestUtil:
  Utility to create Kafka topics before tests.

• Apache Camel Context:
  Uses Camel's RouteBuilder, ProducerTemplate, and MockEndpoint to define routes and verify message flow.

• SimpleIdempotentTest:
  Base class likely providing common test setup and context extensions.

Usage Scenario Summary

1. Setup:
   Create a Kafka topic as the idempotent repository store.

2. Route Configuration:
   Messages sent to "direct:in" are first recorded at "mock:before".
   Then the idempotent consumer checks if the message ID was seen before using the Kafka repository.

3. Message Flow:

   • If unique, message proceeds to "mock:out".

   • If duplicate, message is filtered out and not sent to "mock:out".

4. Testing:
   Sends multiple messages with duplicate IDs and verifies filtering.
   Simulates exceptions to test rollback and idempotency consistency.

Visual Diagram

classDiagram
    class KafkaIdempotentRepositoryEagerIT {
        -REPOSITORY_TOPIC: String
        -idempotentRepository: KafkaIdempotentRepository
        +createRepositoryTopic(): void
        +createRouteBuilder(): RouteBuilder
        +testRemovesDuplicates(): void
        +testRollsBackOnException(): void
    }

    class KafkaIdempotentRepository {
        +KafkaIdempotentRepository(topic: String, bootstrapServers: String)
        +add(key: String): boolean
        +contains(key: String): boolean
    }

    class RouteBuilder {
        +configure(): void
    }

    KafkaIdempotentRepositoryEagerIT --> KafkaIdempotentRepository : uses
    KafkaIdempotentRepositoryEagerIT --> RouteBuilder : creates route with idempotentConsumer

Summary

`KafkaIdempotentRepositoryEagerIT.java` is a focused integration test validating the eager use of a Kafka-backed idempotent repository in Apache Camel routes. It confirms that duplicate messages are filtered correctly and that exceptions trigger proper rollback behavior. This test helps ensure the reliability and correctness of distributed idempotency when using Kafka as a state store in message-driven applications.