KafkaConsumerAuthInvalidIT.java

Overview

`KafkaConsumerAuthInvalidIT.java` is an integration test file within the Apache Camel Kafka component project. Its primary purpose is to verify that the Kafka consumer in Camel correctly handles authentication failures when connecting to a Kafka broker secured with JAAS+SASL (Simple Authentication and Security Layer).

Specifically, the test sets up a Kafka consumer with **invalid SASL credentials** and asserts that:

The consumer fails to authenticate and does not receive messages.
The failed exchanges are routed to a dead letter queue (DLQ).
The Kafka consumer group remains empty, indicating no successful consumer membership.

This test helps ensure that authentication errors are properly detected and handled in Camel Kafka integrations.

Class: `KafkaConsumerAuthInvalidIT`

This class contains JUnit 5 integration tests focused on Kafka consumer authentication failure scenarios.

Annotations and Test Configuration

@EnabledIfSystemProperties: Enables tests only if Kafka version 3.x is used (local-kafka3-container or kafka).
@DisabledIfSystemProperty: Disables tests on CI environments due to flakiness.
@TestMethodOrder(MethodOrderer.OrderAnnotation.class): Orders tests and extensions by @Order.
Uses @RegisterExtension to set up Kafka and Camel test infrastructure.

Constants and Fields

Name	Type	Description
`TOPIC`	`String`	Kafka topic name `"test-auth-invalid-it"`
`kafkaAdminClient`	`AdminClient`	Kafka admin client for managing topics/groups
`LOG`	`Logger`	Logger instance
`service`	`KafkaService` (registered extension)	Kafka service container
`contextExtension`	`CamelContextExtension` (registered extension)	Camel context for route and endpoint management
`producer`	KafkaProducer	Kafka producer used to send messages

Lifecycle Methods

`before()`

Purpose: Initializes Kafka producer with valid SASL credentials and clears any captured records in the mock interceptor.
Implementation:
- Obtains default Kafka properties from the service.
- Sets SASL JAAS config with valid credentials (camel / camel-secret).
- Configures security protocol and SASL mechanism (SASL_PLAINTEXT and PLAIN).
- Instantiates a KafkaProducer.
Throws: Fails the test if producer creation throws an exception.

`setKafkaAdminClient()`

Purpose: Lazily initializes the Kafka AdminClient used to query consumer group information.
Implementation: Uses KafkaTestUtil.createAdminClient(service).

`after()`

Purpose: Cleans up resources after each test.
Implementation:
- Closes the Kafka producer if it exists.
- Deletes the test topic to cleanup Kafka state.

Route Setup

`createRouteBuilder(CamelContext context)`

Purpose: Registers a Camel route defined by createRouteBuilder().
Usage: Used by Camel test infra to set up routes before tests run.

`createRouteBuilder() : RouteBuilder`

Purpose: Defines a Kafka consumer route configured with invalid SASL credentials.
Details:
- The SASL JAAS config uses username "camel" but an invalid password "camel-invalid-secret".
- The route consumes from the configured TOPIC with:
  - groupId: "KafkaConsumerAuthInvalidIT"
  - autoOffsetReset=earliest
  - Deserializers for String key and value.
  - SASL and security protocol settings.
- Messages are processed by logging the body at trace level.
- Dead letter channel error handler routes failures to "mock:dlq".
- Successfully consumed messages would be sent to a mock endpoint "mock:result" (but none should be received due to auth failure).
Route ID: "should-no-work"

Test Method

`kafkaMessageIsConsumedByCamel()`

Annotations:
- @DisplayName("Tests that Camel can adequately handle invalid authorizations")
- @Timeout(30)
- @Test
Purpose: Verifies that the consumer fails due to invalid authentication and messages are routed to the DLQ.
Test Flow:
1. Retrieve mock endpoints for result and dead letter queue.
2. Expect 1 message in the DLQ and 0 in the result endpoint.
3. Assert expectations within 3000 ms.
4. Query the Kafka consumer group info to verify there are no members.
5. Assert each message in DLQ originated from the route "should-no-work".
Assertions:
- DLQ receives 1 failed exchange.
- Result endpoint receives no messages.
- Consumer group "KafkaConsumerAuthInvalidIT" has zero members.
- The failed message originates from the expected route.

Important Implementation Details

Authentication Failure Simulation: The test deliberately uses an invalid SASL JAAS configuration ("camel-invalid-secret") to simulate authentication failure to Kafka.
Dead Letter Queue: Camel's error handler is configured with a deadLetterChannel("mock:dlq") to capture failed exchanges caused by authentication errors.
Kafka AdminClient: Used to inspect the Kafka consumer group state, proving no consumers joined due to auth failure.
System Property Conditions: The test only runs on Kafka 3.x instances and is disabled on CI environments to avoid flaky failures.

Interaction with Other Components

KafkaTestUtil & KafkaAdminUtil: Utility classes from the integration test suite used for Kafka client configuration and admin operations.
ContainerLocalAuthKafkaService: Provides SASL JAAS configuration strings and manages Kafka container lifecycle with authentication enabled.
CamelContext and Routes: The test adds a route to the Camel context that consumes from Kafka using invalid authentication.
Mock Endpoints: Used to verify message routing outcomes inside Camel routes.
KafkaServiceFactory: Manages Kafka container instances used for testing.

Usage Example

This is a test class and thus is run by a test runner (e.g., Maven Surefire or IDE JUnit runner). To execute the test:

Ensure the system properties specify Kafka 3.x:
- -Dkafka.instance.type=local-kafka3-container
Run the test via Maven or IDE.
The test will:
- Setup Kafka container with SASL+JAAS authentication.
- Start a Camel route consuming Kafka with invalid SASL credentials.
- Verify the consumer fails to authenticate and routes failures to DLQ.

Mermaid Class Diagram

classDiagram
    class KafkaConsumerAuthInvalidIT {
        <<test class>>
        +static final String TOPIC
        -static AdminClient kafkaAdminClient
        -static final Logger LOG
        -static KafkaService service
        -static CamelContextExtension contextExtension
        -KafkaProducer<String,String> producer
        +void before()
        +void setKafkaAdminClient()
        +void after()
        +void createRouteBuilder(CamelContext)
        +RouteBuilder createRouteBuilder()
        +void kafkaMessageIsConsumedByCamel()
    }

Summary

`KafkaConsumerAuthInvalidIT.java` is a focused integration test validating that Apache Camel's Kafka consumer properly handles invalid SASL authentication by failing to consume messages and routing failed exchanges to a dead letter queue. It leverages Kafka test infrastructure, Camel routing, and JUnit 5 features to automate and verify this behavior in a controlled test Kafka environment.