kafka-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Jason Gustafson (JIRA)" <j...@apache.org>
Subject [jira] [Resolved] (KAFKA-4491) OOME when Java client can not connect to brokers
Date Thu, 19 Jan 2017 21:52:26 GMT

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

Jason Gustafson resolved KAFKA-4491.
------------------------------------
    Resolution: Duplicate

Resolving this in favor KAFKA-4493, which has a patch available.

> OOME when Java client can not connect to brokers
> ------------------------------------------------
>
>                 Key: KAFKA-4491
>                 URL: https://issues.apache.org/jira/browse/KAFKA-4491
>             Project: Kafka
>          Issue Type: Bug
>          Components: clients
>    Affects Versions: 0.10.0.1
>         Environment: Any (Linux|Windows)
>            Reporter: GĂ©rald Quintana
>
> Scenario: The broker cluster switched to SSL protocol but not the clients, this should
have raised connection failures, but instead the client dies with OOME.
> Sample code to easily reproce the problem:
> {code}
> public class Main {
>     private static final Logger LOGGER = LoggerFactory.getLogger(Main.class);
>     public static void main(String[] args) throws InterruptedException {
>         int threads = 10;
>         ExecutorService executorService = Executors.newFixedThreadPool(threads);
>         for (int i = 0; i < threads; i++) {
>             executorService.execute(new PrintConsumer("testgroup"+i, "testtopic"+i));
>         }
>         Thread.sleep(300000L);
>         executorService.shutdown();
>     }
>     private static class PrintConsumer implements Runnable {
>         private final String groupId;
>         private final String topic;
>         private final AtomicBoolean running = new AtomicBoolean(true);
>         public PrintConsumer(String groupId, String topic) {
>             this.groupId = groupId;
>             this.topic = topic;
>         }
>         @Override
>         public void run() {
>             Properties props = new Properties();
>             props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:9093,kafka2:9093");
>             props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
>             props.put(ConsumerConfig.CLIENT_ID_CONFIG,"testclient");
>             props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
>             props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
>             //props.put("ssl.truststore.location","/opt/truststore.jks");
>             //props.put("ssl.truststore.password", "localhost");
>             //props.put("security.protocol", "SSL");
>             while (running.get()) {
>                 LOGGER.info("Connecting "+topic);
>                 try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props))
{
>                     consumer.subscribe(Collections.singleton(topic));
>                     while (running.get()) {
>                         ConsumerRecords<String, String> records = consumer.poll(100);
>                         for (ConsumerRecord<String, String> record : records)
>                             LOGGER.info("offset = %d, key = %s, value = %s%n", record.offset(),
record.key(), record.value());
>                     }
>                 } catch (Exception e) {
>                     LOGGER.warn("Exception in "+groupId, e);
>                 }
>             }
>         }
>         public void stop() {
>             running.set(false);
>         }
>     }
> }
> {code}
> Thrown exception:
> {code}
> java.lang.OutOfMemoryError: Java heap space
> 	at java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:57)
> 	at java.nio.ByteBuffer.allocate(ByteBuffer.java:335)
> 	at org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:93)
> 	at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71)
> 	at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:154)
> 	at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:135)
> 	at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:323)
> 	at org.apache.kafka.common.network.Selector.poll(Selector.java:283)
> 	at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:260)
> 	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:360)
> 	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:224)
> 	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:192)
> 	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:163)
> 	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:179)
> 	at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:973)
> 	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:937)
> 	at Main$PrintConsumer.run(Main.java:51)
> 	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> 	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> {code}
> Increasing heap size doesn't help, the Kafka client swallows everything.
> As soon as you fix configuration (uncomment SSL properties in configuration), everything
is alright.
> The allocation path seems to be:
> {code} 
> this     - value: byte[] #1
>  <- hb (Java frame)     - class: java.nio.HeapByteBuffer, value: byte[] #1
>   <- buffer (Java frame)     - class: org.apache.kafka.common.network.NetworkReceive,
value: java.nio.HeapByteBuffer #5
>    <- receive (Java frame)     - class: org.apache.kafka.common.network.KafkaChannel,
value: org.apache.kafka.common.network.NetworkReceive #5
>     <- attachment (Java frame)     - class: sun.nio.ch.SelectionKeyImpl, value: org.apache.kafka.common.network.KafkaChannel
#5
> {code}



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

Mime
View raw message