You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Mickael Maison (Jira)" <ji...@apache.org> on 2019/11/07 14:37:00 UTC
[jira] [Commented] (KAFKA-8933) An unhandled SSL handshake
exception in polling event - needed a retry logic
[ https://issues.apache.org/jira/browse/KAFKA-8933?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16969309#comment-16969309 ]
Mickael Maison commented on KAFKA-8933:
---------------------------------------
We saw a similar exception in 2.4/2.5:
{code:java}
Nov 5 18:10:26 mirrormaker2-6c5bbffffc-jx85h mirrormaker2 ERROR [MirrorSourceConnector|task-0] Failure during poll. (org.apache.kafka.connect.mirror.MirrorSourceTask:159)
java.lang.NullPointerException
at org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.handleCompletedMetadataResponse(NetworkClient.java:1071)
at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:847)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:549)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:262)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.pollNoWakeup(ConsumerNetworkClient.java:303)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1249)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1211)
at org.apache.kafka.connect.mirror.MirrorSourceTask.poll(MirrorSourceTask.java:137)
at org.apache.kafka.connect.runtime.WorkerSourceTask.poll(WorkerSourceTask.java:259)
at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:226)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:177)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:227)
at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
at java.util.concurrent.FutureTask.run(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
at java.lang.Thread.run(Unknown Source){code}
Prior to the exception, the client was disconnected. In NetworkClient.processDisconnection() several paths can lead to inProgressRequestVersion being set to null. If there's a MetadataRequest in flight to another node at the time, then the exception is hit when handling the response as inProgressRequestVersion is unboxed to an int.
I was able to reproduce reliably with the following setup:
- 2 brokers with SASL
- consumer consuming from broker 0
- make broker 0 drop consumer connection
- the consumer gets:
{code:java}
INFO [main]: [Consumer clientId=consumer-2, groupId=null] Error sending fetch request (sessionId=735517, epoch=2) to node 0: {}.
org.apache.kafka.common.errors.DisconnectException{code}
- make broker 0 fail authentication
- consumer gets:
{code:java}
INFO [main]: [Consumer clientId=consumer-2, groupId=null] Failed authentication with localhost/127.0.0.1 (Authentication failed, invalid credentials)
ERROR [main]: [Consumer clientId=consumer-2, groupId=null] Connection to node 0 (localhost/127.0.0.1:9093) failed authentication due to: Authentication failed, invalid credentials{code}
- force metadata refresh by consumer, for example: consumer.partitionsFor("topic-that-does-not-exist");
- the consumer gets:
{code:java}
java.lang.NullPointerException
at org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.handleCompletedMetadataResponse(NetworkClient.java:1073)
at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:847)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:549)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:262)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
at org.apache.kafka.clients.consumer.internals.Fetcher.getTopicMetadata(Fetcher.java:368)
at org.apache.kafka.clients.consumer.KafkaConsumer.partitionsFor(KafkaConsumer.java:1930)
at org.apache.kafka.clients.consumer.KafkaConsumer.partitionsFor(KafkaConsumer.java:1898)
at main.ConsumerTest2.main(ConsumerTest2.java:37)
{code}
I'm not super familair with this code path but the following patch helped:
{code:java}
diff --git clients/src/main/java/org/apache/kafka/clients/NetworkClient.java clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
index d782df865..d3119f132 100644
--- clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
+++ clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -1067,6 +1069,9 @@ public class NetworkClient implements KafkaClient {
if (response.brokers().isEmpty()) {
log.trace("Ignoring empty metadata response with correlation id {}.", requestHeader.correlationId());
this.metadata.failedUpdate(now, null);
+ } else if (inProgressRequestVersion == null) {
+ log.warn("Ignoring metadata response ...");
+ this.metadata.failedUpdate(now, null);
} else {
this.metadata.update(inProgressRequestVersion, response, now);
}
{code}
> An unhandled SSL handshake exception in polling event - needed a retry logic
> ----------------------------------------------------------------------------
>
> Key: KAFKA-8933
> URL: https://issues.apache.org/jira/browse/KAFKA-8933
> Project: Kafka
> Issue Type: Bug
> Components: clients
> Affects Versions: 2.2.1
> Environment: software platform
> Reporter: Remigius
> Priority: Critical
>
> Already client is connected and during polling event, SSL handshake failure happened. it led to leaving the co-ordinator. Even on SSL handshake failure which was actually intermittent issue, polling should have some resilient and retry the polling. Leaving group caused all instances of clients to drop and left the messages in Kafka for long time until re-subscribe the kafka topic manually.
>
>
> {noformat}
> 2019-09-06 04:03:09,016 ERROR [reactive-kafka-xxxx] org.apache.kafka.clients.NetworkClient [Consumer clientId=aaa, groupId=bbb] Connection to node 150 (host:port) failed authentication due to: SSL handshake failed
> 2019-09-06 04:03:09,021 ERROR [reactive-kafka-xxxx] reactor.kafka.receiver.internals.DefaultKafkaReceiver Unexpected exception
> java.lang.NullPointerException: null
> at org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.handleCompletedMetadataResponse(NetworkClient.java:1012) ~[kafka-clients-2.2.1.jar!/:?]
> at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:822) ~[kafka-clients-2.2.1.jar!/:?]
> at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:544) ~[kafka-clients-2.2.1.jar!/:?]
> at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:265) ~[kafka-clients-2.2.1.jar!/:?]
> at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:236) ~[kafka-clients-2.2.1.jar!/:?]
> at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1256) ~[kafka-clients-2.2.1.jar!/:?]
> at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1200) ~[kafka-clients-2.2.1.jar!/:?]
> at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1176) ~[kafka-clients-2.2.1.jar!/:?]
> at reactor.kafka.receiver.internals.DefaultKafkaReceiver$PollEvent.run(DefaultKafkaReceiver.java:470) ~[reactor-kafka-1.1.1.RELEASE.jar!/:1.1.1.RELEASE]
> at reactor.kafka.receiver.internals.DefaultKafkaReceiver.doEvent(DefaultKafkaReceiver.java:401) ~[reactor-kafka-1.1.1.RELEASE.jar!/:1.1.1.RELEASE]
> at reactor.kafka.receiver.internals.DefaultKafkaReceiver.lambda$start$14(DefaultKafkaReceiver.java:335) ~[reactor-kafka-1.1.1.RELEASE.jar!/:1.1.1.RELEASE]
> at reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:130) ~[reactor-core-3.2.10.RELEASE.jar!/:3.2.10.RELEASE]
> at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.runAsync(FluxPublishOn.java:398) ~[reactor-core-3.2.10.RELEASE.jar!/:3.2.10.RELEASE]
> at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.run(FluxPublishOn.java:484) ~[reactor-core-3.2.10.RELEASE.jar!/:3.2.10.RELEASE]
> at reactor.kafka.receiver.internals.KafkaSchedulers$EventScheduler.lambda$decorate$1(KafkaSchedulers.java:100) ~[reactor-kafka-1.1.1.RELEASE.jar!/:1.1.1.RELEASE]
> at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) ~[reactor-core-3.2.10.RELEASE.jar!/:3.2.10.RELEASE]
> at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) ~[reactor-core-3.2.10.RELEASE.jar!/:3.2.10.RELEASE]
> at org.springframework.cloud.sleuth.instrument.async.TraceCallable.call(TraceCallable.java:70) ~[spring-cloud-sleuth-core-2.1.1.RELEASE.jar!/:2.1.1.RELEASE]
> at java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?]
> at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) ~[?:?]
> at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[?:?]
> at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[?:?]
> at java.lang.Thread.run(Thread.java:834) [?:?]
> 2019-09-06 04:03:09,023 INFO [reactive-kafka-xxxx] org.apache.kafka.clients.consumer.internals.AbstractCoordinator [Consumer clientId=aaa, groupId=bbb] Member x_13-081e61ec-1509-4e0e-819e-58063d1ce8f6 sending LeaveGroup request to coordinator{noformat}
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)