Mailing-List: contact dev-help@kafka.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@kafka.apache.org
Date: Thu, 19 Jan 2017 21:52:26 +0000 (UTC)
From: "Jason Gustafson (JIRA)" <jira@apache.org>
To: dev@kafka.apache.org
Message-ID: <JIRA.13025928.1481021533000.56567.1484862746654@Atlassian.JIRA>
In-Reply-To: <JIRA.13025928.1481021533000@Atlassian.JIRA>
References: <JIRA.13025928.1481021533000@Atlassian.JIRA> <JIRA.13025928.1481021533130@jira-lw-us.apache.org>
Subject: [jira] [Resolved] (KAFKA-4491) OOME when Java client can not
 connect to brokers
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: quoted-printable
archived-at: Thu, 19 Jan 2017 21:52:36 -0000


     [ https://issues.apache.org/jira/browse/KAFKA-4491?page=3Dcom.atlassia=
n.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=C3=A9rald 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 =3D LoggerFactory.getLogger(Main.c=
lass);
>     public static void main(String[] args) throws InterruptedException {
>         int threads =3D 10;
>         ExecutorService executorService =3D Executors.newFixedThreadPool(=
threads);
>         for (int i =3D 0; i < threads; i++) {
>             executorService.execute(new PrintConsumer("testgroup"+i, "tes=
ttopic"+i));
>         }
>         Thread.sleep(300000L);
>         executorService.shutdown();
>     }
>     private static class PrintConsumer implements Runnable {
>         private final String groupId;
>         private final String topic;
>         private final AtomicBoolean running =3D new AtomicBoolean(true);
>         public PrintConsumer(String groupId, String topic) {
>             this.groupId =3D groupId;
>             this.topic =3D topic;
>         }
>         @Override
>         public void run() {
>             Properties props =3D new Properties();
>             props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:90=
93,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, "or=
g.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 =3D new Kafka=
Consumer<>(props)) {
>                     consumer.subscribe(Collections.singleton(topic));
>                     while (running.get()) {
>                         ConsumerRecords<String, String> records =3D consu=
mer.poll(100);
>                         for (ConsumerRecord<String, String> record : reco=
rds)
>                             LOGGER.info("offset =3D %d, key =3D %s, value=
 =3D %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
> =09at java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:57)
> =09at java.nio.ByteBuffer.allocate(ByteBuffer.java:335)
> =09at org.apache.kafka.common.network.NetworkReceive.readFromReadableChan=
nel(NetworkReceive.java:93)
> =09at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkRece=
ive.java:71)
> =09at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.j=
ava:154)
> =09at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java=
:135)
> =09at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector=
.java:323)
> =09at org.apache.kafka.common.network.Selector.poll(Selector.java:283)
> =09at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:260)
> =09at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.c=
lientPoll(ConsumerNetworkClient.java:360)
> =09at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.p=
oll(ConsumerNetworkClient.java:224)
> =09at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.p=
oll(ConsumerNetworkClient.java:192)
> =09at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.p=
oll(ConsumerNetworkClient.java:163)
> =09at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ens=
ureCoordinatorReady(AbstractCoordinator.java:179)
> =09at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsu=
mer.java:973)
> =09at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.=
java:937)
> =09at Main$PrintConsumer.run(Main.java:51)
> =09at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecuto=
r.java:1142)
> =09at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecut=
or.java:617)
> {code}
> Increasing heap size doesn't help, the Kafka client swallows everything.
> As soon as you fix configuration (uncomment SSL properties in configurati=
on), everything is alright.
> The allocation path seems to be:
> {code}=20
> this     - value: byte[] #1
>  <- hb (Java frame)     - class: java.nio.HeapByteBuffer, value: byte[] #=
1
>   <- buffer (Java frame)     - class: org.apache.kafka.common.network.Net=
workReceive, value: java.nio.HeapByteBuffer #5
>    <- receive (Java frame)     - class: org.apache.kafka.common.network.K=
afkaChannel, 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)