Mailing-List: contact dev-help@kafka.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@kafka.apache.org
Date: Thu, 1 Jun 2017 20:56:04 +0000 (UTC)
From: "Ismael Juma (JIRA)" <jira@apache.org>
To: dev@kafka.apache.org
Message-ID: <JIRA.13075612.1496064872000.345060.1496350564429@Atlassian.JIRA>
In-Reply-To: <JIRA.13075612.1496064872000@Atlassian.JIRA>
References: <JIRA.13075612.1496064872000@Atlassian.JIRA> <JIRA.13075612.1496064872138@jira-lw-us.apache.org>
Subject: [jira] [Resolved] (KAFKA-5345) Some socket connections not closed
 after restart of Kafka Streams
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit
archived-at: Thu, 01 Jun 2017 20:56:11 -0000


     [ https://issues.apache.org/jira/browse/KAFKA-5345?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Ismael Juma resolved KAFKA-5345.
--------------------------------
       Resolution: Fixed
    Fix Version/s: 0.11.0.0

Issue resolved by pull request 3195
[https://github.com/apache/kafka/pull/3195]

> Some socket connections not closed after restart of Kafka Streams
> -----------------------------------------------------------------
>
>                 Key: KAFKA-5345
>                 URL: https://issues.apache.org/jira/browse/KAFKA-5345
>             Project: Kafka
>          Issue Type: Bug
>          Components: streams
>    Affects Versions: 0.10.2.0, 0.10.2.1
>         Environment: MacOs 10.12.5 and Ubuntu 14.04
>            Reporter: Jeroen van Wilgenburg
>            Assignee: Rajini Sivaram
>             Fix For: 0.11.0.0
>
>
> We ran into a problem that resulted in a "Too many open files" exception because some sockets are not closed after a restart.
> This problem only occurs with version {{0.10.2.1}} and {{0.10.2.0}}. 
> {{0.10.1.1}} and {{0.10.1.0}} both work as expected.
> I used the same version for the server and client.
> I used https://github.com/kohsuke/file-leak-detector to display the open file descriptors. The culprit was :
> {noformat}
> #146 socket channel by thread:pool-2-thread-1 on Mon May 29 11:20:47 CEST 2017
> 	at sun.nio.ch.SocketChannelImpl.<init>(SocketChannelImpl.java:108)
> 	at sun.nio.ch.SelectorProviderImpl.openSocketChannel(SelectorProviderImpl.java:60)
> 	at java.nio.channels.SocketChannel.open(SocketChannel.java:145)
> 	at org.apache.kafka.common.network.Selector.connect(Selector.java:168)
> 	at org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:629)
> 	at org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:186)
> 	at org.apache.kafka.streams.processor.internals.StreamsKafkaClient.ensureOneNodeIsReady(StreamsKafkaClient.java:195)
> 	at org.apache.kafka.streams.processor.internals.StreamsKafkaClient.getAnyReadyBrokerId(StreamsKafkaClient.java:233)
> 	at org.apache.kafka.streams.processor.internals.StreamsKafkaClient.checkBrokerCompatibility(StreamsKafkaClient.java:300)
> 	at org.apache.kafka.streams.KafkaStreams.checkBrokerVersionCompatibility(KafkaStreams.java:401)
> 	at org.apache.kafka.streams.KafkaStreams.start(KafkaStreams.java:425)
> {noformat}	
> 	
> 	
> I could narrow the problem down to a reproducable example below (the only dependency is 
> {{org.apache.kafka:kafka-streams:jar:0.10.2.1}}). 
> *IMPORTANT*: You have to run this code in the Intellij IDEA debugger with a special breakpoint to see it fail. 
> See the comments on the socketChannels variable on how to add this breakpoint. 
> When you run this code you will see the number of open SocketChannels increase (only on version 0.10.2.x).
> 	
> {code:title=App.java}
> import org.apache.kafka.common.serialization.Serdes;
> import org.apache.kafka.streams.KafkaStreams;
> import org.apache.kafka.streams.StreamsConfig;
> import org.apache.kafka.streams.kstream.KStream;
> import org.apache.kafka.streams.kstream.KStreamBuilder;
> import java.nio.channels.SocketChannel;
> import java.nio.channels.spi.AbstractInterruptibleChannel;
> import java.util.ArrayList;
> import java.util.List;
> import java.util.Properties;
> import java.util.concurrent.Executors;
> import java.util.concurrent.ScheduledExecutorService;
> import java.util.concurrent.TimeUnit;
> import java.util.stream.Collectors;
> public class App {
>     private static KafkaStreams streams;
>     private static String brokerList;
>     // Fill socketChannels with entries on line 'Socket socket = socketChannel.socket();' (line number 170  on 0.10.2.1)
>     // of org.apache.kafka.common.network.Selector: Add breakpoint, right click on breakpoint.
>     // - Uncheck 'Suspend'
>     // - Check 'Evaluate and log' and fill text field with (without quotes) 'App.socketChannels.add(socketChannel)'
>     private static final List<SocketChannel> socketChannels = new ArrayList<>();
>     public static void main(String[] args) {
>         brokerList = args[0];
>         init();
>         ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(1);
>         Runnable command = () -> {
>             streams.close();
>             System.out.println("Open socketChannels: " + socketChannels.stream()
>                     .filter(AbstractInterruptibleChannel::isOpen)
>                     .collect(Collectors.toList()).size());
>             init();
>         };
>         scheduledThreadPool.scheduleWithFixedDelay(command, 10000L, 2000, TimeUnit.MILLISECONDS);
>     }
>     private static void init() {
>         Properties streamsConfiguration = new Properties();
>         streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "JeroenApp");
>         streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
>         StreamsConfig config = new StreamsConfig(streamsConfiguration);
>         KStreamBuilder builder = new KStreamBuilder();
>         KStream<String, String> stream = builder.stream(Serdes.String(), Serdes.String(), "HarrieTopic");
>         stream.foreach((key, value) -> System.out.println(value));
>         streams = new KafkaStreams(builder, config);
>         streams.start();
>     }
> }
> {code}


--
This message was sent by Atlassian JIRA
(v6.3.15#6346)