[
https://issues.apache.org/jira/browse/FLINK-9009?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Pankaj reopened FLINK-9009:
---------------------------
Above Issue only comes when we use CassandraSink of flink.
If we use flink parallelism =10 and try to write stream events from kafka topic
to cassandra, system throws error:
*You are creating too many HashedWheelTimer instances. HashedWheelTimer is a
shared resource that must be reused across the application, so that only a few
instances are created-> Also reason for OOM of JVM*
For your references: It seems Flink CassandraSink is not closing cassandra
session when we increase parallelism.
One can easily replicate the issue :
step1: CPU core 200 multicore
step2: Kafkatopic
step3: parallelism=10
step4: memory 512MB
Classes : CassandraSink
Snippet:
public class App {
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(20);
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("group.id", "test");
FlinkKafkaConsumer011<String> myConsumer = new
FlinkKafkaConsumer011<>("test", new SimpleStringSchema(),
properties);
DataStream<AtomicEvent> stream = env.addSource(myConsumer).map(new
MapFunction<String, AtomicEvent>() {
private static final long serialVersionUID = 1L;
@Override
public AtomicEvent map(String value) throws Exception {
return new AtomicEvent("xyx", new Date(), "1",
ByteBuffer.wrap("a".getBytes()), "1");
}
});
CassandraSink.addSink(stream).setClusterBuilder(new ClusterBuilder() {
private static final long serialVersionUID = 1L;
@Override
public Cluster buildCluster(Cluster.Builder builder) {
return builder.addContactPoint("127.0.0.1").build();
}
}).build();
env.execute();
}
}
Mar 19, 2018 6:16:19 PM com.datastax.driver.core.NettyUtil <clinit>
INFO: Did not find Netty's native epoll transport in the classpath, defaulting
to NIO.
Mar 19, 2018 6:16:19 PM io.netty.util.internal.logging.Slf4JLogger error
SEVERE: LEAK: You are creating too many HashedWheelTimer instances.
HashedWheelTimer is a shared resource that must be reused across the JVM,so
that only a few instances are created.
Seems some issue at flink CassandraSinkBase implementation
> Error| You are creating too many HashedWheelTimer instances.
> HashedWheelTimer is a shared resource that must be reused across the
> application, so that only a few instances are created.
> -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-9009
> URL: https://issues.apache.org/jira/browse/FLINK-9009
> Project: Flink
> Issue Type: Bug
> Environment: Pass platform: Openshit
> Reporter: Pankaj
> Priority: Major
>
> Steps to reproduce:
> 1- Flink with Kafka as a consumer -> Writing stream to Cassandra using flink
> cassandra sink.
> 2- In memory Job manager and task manager with checkpointing 5000ms.
> 3- env.setpararllelism(10)-> As kafka topic has 10 partition.
> 4- There are around 13 unique streams in a single flink run time environment
> which are reading from kafka -> processing and writing to cassandra.
> Hardware: CPU 200 milli core . It is deployed on Paas platform on one node
> Memory: 526 MB.
>
> When i start the server, It starts flink and all off sudden stops with above
> error. It also shows out of memory error.
>
> It would be nice if any body can suggest if something is wrong.
>
> Maven:
> flink-connector-cassandra_2.11: 1.3.2
> flink-streaming-java_2.11: 1.4.0
> flink-connector-kafka-0.11_2.11:1.4.0
>
>
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
