[jira] [Comment Edited] (KAFKA-2672) SendFailedException when new consumer is run with SSL

["Ismael Juma (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-2672?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14965331#comment-14965331 ] 

Ismael Juma edited comment on KAFKA-2672 at 10/20/15 4:22 PM:
--------------------------------------------------------------

[~junrao], that is right. As far as I can see, it does not:

{code}
    private boolean trySend(long now) {
        // send any requests that can be sent now
        boolean requestsSent = false;
        for (Map.Entry<Node, List<ClientRequest>> requestEntry: unsent.entrySet()) {
            Node node = requestEntry.getKey();
            Iterator<ClientRequest> iterator = requestEntry.getValue().iterator();
            while (iterator.hasNext()) {
                ClientRequest request = iterator.next();
                if (client.ready(node, now)) {
                    client.send(request, now);
                    iterator.remove();
                    requestsSent = true;
                } else if (client.connectionFailed(node)) {
                    RequestFutureCompletionHandler handler =
                            (RequestFutureCompletionHandler) request.callback();
                    handler.onComplete(new ClientResponse(request, now, true, null));
                    iterator.remove();
                }
            }
        }
        return requestsSent;
    }
{code}

The log entry that Rajini quoted is triggered by an exception raised in `clearUnsentRequests`:

{code}
private void clearUnsentRequests(long now) {
        // clear all unsent requests and fail their corresponding futures
        for (Map.Entry<Node, List<ClientRequest>> requestEntry: unsent.entrySet()) {
            Iterator<ClientRequest> iterator = requestEntry.getValue().iterator();
            while (iterator.hasNext()) {
                ClientRequest request = iterator.next();
                RequestFutureCompletionHandler handler =
                        (RequestFutureCompletionHandler) request.callback();
                handler.raise(SendFailedException.INSTANCE);
                iterator.remove();
            }
        }
        unsent.clear();
    }
{code}

In my original comment, I showed that `clearUnsentRequests` is called before `ConsumerNetworkClient.poll` returns. I haven't investigated in detail, but it sounds like the handshake takes longer to complete than `poll` and hence the log output. I was hoping that Jason would confirm or clarify this. :)


was (Author: ijuma):
[~junrao], that is right. As far as I can see, it does not:

{code}
    private boolean trySend(long now) {
        // send any requests that can be sent now
        boolean requestsSent = false;
        for (Map.Entry<Node, List<ClientRequest>> requestEntry: unsent.entrySet()) {
            Node node = requestEntry.getKey();
            Iterator<ClientRequest> iterator = requestEntry.getValue().iterator();
            while (iterator.hasNext()) {
                ClientRequest request = iterator.next();
                if (client.ready(node, now)) {
                    client.send(request, now);
                    iterator.remove();
                    requestsSent = true;
                } else if (client.connectionFailed(node)) {
                    RequestFutureCompletionHandler handler =
                            (RequestFutureCompletionHandler) request.callback();
                    handler.onComplete(new ClientResponse(request, now, true, null));
                    iterator.remove();
                }
            }
        }
        return requestsSent;
    }
{code}

The log entry that Rajini quoted happens in `clearUnsentRequests`:

{code}
private void clearUnsentRequests(long now) {
        // clear all unsent requests and fail their corresponding futures
        for (Map.Entry<Node, List<ClientRequest>> requestEntry: unsent.entrySet()) {
            Iterator<ClientRequest> iterator = requestEntry.getValue().iterator();
            while (iterator.hasNext()) {
                ClientRequest request = iterator.next();
                RequestFutureCompletionHandler handler =
                        (RequestFutureCompletionHandler) request.callback();
                handler.raise(SendFailedException.INSTANCE);
                iterator.remove();
            }
        }
        unsent.clear();
    }
{code}

In my original comment, I showed that `clearUnsentRequests` is called before `ConsumerNetworkClient.poll` returns. I haven't investigated in detail, but it sounds like the handshake takes longer to complete than `poll` and hence the log output. I was hoping that Jason would confirm or clarify this. :)

> SendFailedException when new consumer is run with SSL
> -----------------------------------------------------
>
>                 Key: KAFKA-2672
>                 URL: https://issues.apache.org/jira/browse/KAFKA-2672
>             Project: Kafka
>          Issue Type: Bug
>          Components: consumer
>            Reporter: Rajini Sivaram
>            Assignee: Neha Narkhede
>             Fix For: 0.9.0.0
>
>
> When running new consumer with SSL, debug logs show these exceptions every time:
> {quote}
> [2015-10-19 20:57:43,389] DEBUG Fetch failed (org.apache.kafka.clients.consumer.internals.Fetcher)
> org.apache.kafka.clients.consumer.internals.SendFailedException 
> {quote}
> The exception occurs  because send is queued before SSL handshake is complete. I am not sure if the exception is harmless, but it will be good to avoid the exception either way since it feels like an exception that exists to handle edge cases like buffer overflow rather than something in a normal code path.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)