Gérald Quintana created KAFKA-4491:
--------------------------------------
Summary: 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)
