KAFKA-10840: Expose authentication failures in KafkaConsumer.poll() #20212
+270
−0
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This change allows applications to catch authentication failures immediately
instead of failing silently. Previously, when SSL certificates expired or
other authentication issues occurred, the consumer would stop fetching data
without clear indication of the underlying problem.
The implementation adds two new exception classes to provide specific error
handling for different authentication failure scenarios. The
CertificateExpiredAuthenticationException is thrown when SSL certificate
expiration is detected in the error message, while
PersistentAuthenticationException handles other non-retriable authentication
failures including SASL errors and general SSL handshake failures.
The ClassicKafkaConsumer.poll() method now actively checks all known cluster
nodes for authentication exceptions before proceeding with fetch operations.
When an authentication failure is detected, detailed error messages including
the affected node information are logged and appropriate exceptions are thrown
to the application layer.
The ConsumerNetworkClient has been enhanced with an authenticationException()
method that exposes the authentication state from the underlying KafkaClient,
allowing the consumer to access this previously hidden information. For testing
purposes, the MockClient class now supports setting specific authentication
exceptions through a new setNodeAuthenticationFailure() method.
Example usage:
```java
try {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
} catch (CertificateExpiredAuthenticationException e) {
log.error("SSL certificate expired: {}", e.getMessage());
// Handle certificate renewal
} catch (PersistentAuthenticationException e) {
log.error("Authentication failed: {}", e.getMessage());
// Handle authentication configuration
}
```
This addresses the core issue where "data flow stops without indication" when
authentication fails, enabling applications to detect and handle these failures
proactively rather than experiencing silent timeouts.
These test cases validate the authentication failure detection and exception handling implementation added for KAFKA-10840. The tests ensure that the ConsumerNetworkClient correctly exposes authentication exceptions from the underlying KafkaClient through the new authenticationException() method. The AuthenticationExceptionTypesTest verifies that the new exception classes CertificateExpiredAuthenticationException and PersistentAuthenticationException maintain proper inheritance hierarchy and behave correctly. These exception types allow applications to distinguish between certificate expiration scenarios and other persistent authentication failures. Additional tests have been added to the existing ConsumerNetworkClientTest to verify that authentication exceptions are properly exposed when present and return null when no authentication issues exist. All tests pass and confirm that the implementation correctly addresses the silent authentication failure problem described in the original JIRA issue.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Summary
This PR implements KAFKA-10840 to expose authentication failures in the KafkaConsumer.poll() method, allowing applications to catch authentication issues immediately instead of experiencing silent failures.
Problem
Previously, when SSL certificates expired or other authentication issues occurred, the consumer would stop fetching data without clear indication of the underlying problem. This led to "data flow stops without indication" scenarios that were difficult to troubleshoot and handle gracefully.
Solution
New Exception Classes
Core Changes
ClassicKafkaConsumer.poll()to actively check all known cluster nodes for authentication exceptions before proceeding with fetch operationsauthenticationException(Node)method toConsumerNetworkClientto expose authentication state from the underlyingKafkaClientMockClientwithsetNodeAuthenticationFailure()method for testing authentication failure scenariosError Detection Logic
The implementation detects certificate expiration by checking for specific patterns in SSL exception messages:
Usage Example
Test Plan
Files Changed
clients/src/main/java/org/apache/kafka/common/errors/CertificateExpiredAuthenticationException.java(NEW)clients/src/main/java/org/apache/kafka/common/errors/PersistentAuthenticationException.java(NEW)clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java(MODIFIED)clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java(MODIFIED)clients/src/test/java/org/apache/kafka/clients/MockClient.java(MODIFIED)This addresses the core issue where "data flow stops without indication" when authentication fails, enabling applications to detect and handle these failures proactively rather than experiencing silent timeouts.