[jira] [Commented] (FLINK-9009) 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 insta
[ https://issues.apache.org/jira/browse/FLINK-9009?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16413587#comment-16413587 ] Pankaj commented on FLINK-9009: --- CassandraSink when given parallelism=10 at enviornment level. It creates 10 different cluster for a stream. Is there any possibility that all threads can reuse one single cluster object. Cause I checked and found that every thread is executing Line82 of CassandraSinkBase.java this.cluster = builder.getCluster(); So, In Case when you have for an example 5 streams in a single Flink enviornment with parallelism=10 and all those 5 streams are required to be persisted in Cassandra. This would result in 50 cluster object in a single flink run time enviornment. I think all threads which belong to a stream should share the cluster object. This is very important to understand and to solve "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." Please correct if i mistaken anything. > 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: Openshift >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)
[jira] [Commented] (FLINK-9009) 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 insta
[ https://issues.apache.org/jira/browse/FLINK-9009?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16404760#comment-16404760 ] Pankaj commented on FLINK-9009: --- Also, If we remove CassandraSink. Everything works fine. One can also check the same. Increase number of parallelism . One can easily replicate issue. > 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: Blocker > > 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)
[jira] [Updated] (FLINK-9009) 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 instanc
[ https://issues.apache.org/jira/browse/FLINK-9009?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Pankaj updated FLINK-9009: -- Priority: Blocker (was: Major) > 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: Blocker > > 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)
[jira] [Reopened] (FLINK-9009) 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 instan
[
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 myConsumer = new
FlinkKafkaConsumer011<>("test", new SimpleStringSchema(),
properties);
DataStream stream = env.addSource(myConsumer).map(new
MapFunction() {
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
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)
[jira] [Commented] (FLINK-9009) 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 insta
[ https://issues.apache.org/jira/browse/FLINK-9009?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16403461#comment-16403461 ] Pankaj commented on FLINK-9009: --- Stacktrace Sink: Cassandra Sink, Filter -> Flat Map -> Sink: Unnamed, Filter -> Flat Map -> Sink: Cassandra Sink, Filter -> Flat Map -> Sink: Unnamed, Filter -> Flat Map -> Sink: *Cassandra Sink)(1/10*) switched to RUNNING 17.03.2018 14:33:59.740 [OUT] [ERROR] [ ] [ ] *io.netty.util.ResourceLeakDetector 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 create*d. # # java.lang.OutOfMemoryError: Java heap space # -XX:OnOutOfMemoryError="/opt/tomcat/bin/shutdown.sh 5" # Executing /bin/sh -c "/opt/tomcat/bin/shutdown.sh 5"... Mar 17, 2018 2:34:28 PM org.apache.catalina.startup.Catalina stopServer SEVERE: Could not contact localhost:8005. Tomcat may not be running. Mar 17, 2018 2:34:28 PM org.apache.catalina.startup.Catalina stopServer SEVERE: Catalina.stop: java.net.ConnectException: Connection refused (Connection refused) at java.net.PlainSocketImpl.socketConnect(Native Method) at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350) at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206) at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188) at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392) at java.net.Socket.connect(Socket.java:589) at java.net.Socket.connect(Socket.java:538) --- "cluster16-nio-worker-1" #142 prio=5 os_prio=0 tid=0x7f97fc386000 nid=0xbe waiting for monitor entry [0x7f9782e86000] java.lang.Thread.State: BLOCKED (on object monitor) at *com.datastax.driver.core.Connection$10.operationComplete(Connection.java:547) at* com.datastax.driver.core.Connection$10.operationComplete(Connection.java:534) at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:680) at io.netty.util.concurrent.DefaultPromise.notifyLateListener(DefaultPromise.java:621) at io.netty.util.concurrent.DefaultPromise.addListener(DefaultPromise.java:138) at io.netty.channel.DefaultChannelPromise.addListener(DefaultChannelPromise.java:93) at io.netty.channel.DefaultChannelPromise.addListener(DefaultChannelPromise.java:28) at com.datastax.driver.core.Connection$Flusher.run(Connection.java:870) at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:358) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:357) at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:112) at java.lang.Thread.run(Thread.java:748) io.netty.handler.codec.MessageToMessageEncoder.write(MessageToMessageEncoder.java:89) at io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:643) at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:700) at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:636) at io.netty.handler.timeout.IdleStateHandler.write(IdleStateHandler.java:284) at io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:643) at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:700) at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:636) at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:622) at io.netty.channel.DefaultChannelPipeline.write(DefaultChannelPipeline.java:939) at io.netty.channel.AbstractChannel.write(AbstractChannel.java:234) at com.datastax.driver.core.Connection$Flusher.run(Connection.java:870) at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:358) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:357) at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:112) at java.lang.Thread.run(Thread.java:748) > 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: Blocker > > Steps to reproduce: > 1- Flink with Kafka as a consumer -> Writing stream to Cassandra using
[jira] [Comment Edited] (FLINK-9009) 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
[ https://issues.apache.org/jira/browse/FLINK-9009?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16402323#comment-16402323 ] Pankaj edited comment on FLINK-9009 at 3/16/18 6:35 PM: No, Is not related with Kafka. I have already tried and check the problem only occurs when we introduced more parallelism and flink is writing to cassandra with two cluster. Lets say in my case I introduced parallelism =10 coz i have 10 partition in kafka topic. I do not face any problem if i use parallelism=1 and cassandra writing from flink. But it faled with more parallelism Problem can be replicated with steps i shared in description. I'm not sure if flink has the fix of below two tickets in the cassandra connector api i shared https://issues.apache.org/jira/browse/CASSANDRA-11243 https://issues.apache.org/jira/browse/CASSANDRA-10837 was (Author: pmishra01): No, Is not related with Kafka. I have already tried and check the problem only occurs when we introduced more parallelism and flink is writing to cassandra with two cluster. Lets say in my case I introduced parallelism =10 coz i have 10 partition in kafka topic. I do not face any problem with same scenario with no cassandra writing from flink. Problem can be replicated with steps i shared in description. I'm not sure if flink has the fix of below two tickets in the cassandra connector api i shared https://issues.apache.org/jira/browse/CASSANDRA-11243 https://issues.apache.org/jira/browse/CASSANDRA-10837 > 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: Blocker > > 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)
[jira] [Comment Edited] (FLINK-9009) 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
[ https://issues.apache.org/jira/browse/FLINK-9009?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16402323#comment-16402323 ] Pankaj edited comment on FLINK-9009 at 3/16/18 6:34 PM: No, Is not related with Kafka. I have already tried and check the problem only occurs when we introduced more parallelism and flink is writing to cassandra with two cluster. Lets say in my case I introduced parallelism =10 coz i have 10 partition in kafka topic. I do not face any problem with same scenario with no cassandra writing from flink. Problem can be replicated with steps i shared in description. I'm not sure if flink has the fix of below two tickets in the cassandra connector api i shared https://issues.apache.org/jira/browse/CASSANDRA-11243 https://issues.apache.org/jira/browse/CASSANDRA-10837 was (Author: pmishra01): No, Is not related with Kafka. I have already tried and check the problem only occurs when we introduced more parallelism and flink is writing two cassandra with two cluster. Lets say in my case I introduced parallelism =10 coz i have 10 partition in kafka topic. I do not face any problem with same scenario with no cassandra writing from flink. Problem can be replicated with steps i shared in description. I'm not sure if flink has the fix of below two tickets in the cassandra connector api i shared https://issues.apache.org/jira/browse/CASSANDRA-11243 https://issues.apache.org/jira/browse/CASSANDRA-10837 > 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: Blocker > > 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)
[jira] [Commented] (FLINK-9009) 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 insta
[ https://issues.apache.org/jira/browse/FLINK-9009?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16402323#comment-16402323 ] Pankaj commented on FLINK-9009: --- No, Is not related with Kafka. I have already tried and check the problem only occurs when we introduced more parallelism and flink is writing two cassandra with two cluster. Lets say in my case I introduced parallelism =10 coz i have 10 partition in kafka topic. I do not face any problem with same scenario with no cassandra writing from flink. Problem can be replicated with steps i shared in description. I'm not sure if flink has the fix of below two tickets in the cassandra connector api i shared https://issues.apache.org/jira/browse/CASSANDRA-11243 https://issues.apache.org/jira/browse/CASSANDRA-10837 > 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: Blocker > > 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)
[jira] [Created] (FLINK-9009) 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 instanc
Pankaj created FLINK-9009: - Summary: 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 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)
[jira] [Commented] (FLINK-7732) Test instability in Kafka end-to-end test (invalid Kafka offset)
[
https://issues.apache.org/jira/browse/FLINK-7732?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16212639#comment-16212639
]
Pankaj commented on FLINK-7732:
---
Hi All, We also faced same issue when using flink with Kafka Consumer. Please
find error traces. Do we have any fix available ?
java.lang.IllegalArgumentException: Invalid offset: -9156...
java.lang.IllegalArgumentException: Invalid offset: -915623761772
at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.sendOffsetCommitRequest(ConsumerCoordinator.java:687)
~[kafka-clients-.
.
> Test instability in Kafka end-to-end test (invalid Kafka offset)
>
>
> Key: FLINK-7732
> URL: https://issues.apache.org/jira/browse/FLINK-7732
> Project: Flink
> Issue Type: Bug
> Components: Kafka Connector, Tests
>Affects Versions: 1.4.0
>Reporter: Nico Kruber
>Priority: Critical
> Labels: test-stability
>
> In a test run with unrelated changes in the network stack, the Kafa
> end-to-end test was failing with an invalid offset:
> {code}
> 2017-09-28 06:34:10,736 INFO org.apache.kafka.common.utils.AppInfoParser
> - Kafka version : 0.10.2.1
> 2017-09-28 06:34:10,744 INFO org.apache.kafka.common.utils.AppInfoParser
> - Kafka commitId : e89bffd6b2eff799
> 2017-09-28 06:34:14,549 INFO
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator - Discovered
> coordinator testing-gce-f67551b8-8867-41c5-b36e-4196d0f11e30:9092 (id:
> 2147483647 rack: null) for group myconsumer.
> 2017-09-28 06:34:14,573 INFO
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator - Marking
> the coordinator testing-gce-f67551b8-8867-41c5-b36e-4196d0f11e30:9092 (id:
> 2147483647 rack: null) dead for group myconsumer
> 2017-09-28 06:34:14,686 INFO
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator - Discovered
> coordinator testing-gce-f67551b8-8867-41c5-b36e-4196d0f11e30:9092 (id:
> 2147483647 rack: null) for group myconsumer.
> 2017-09-28 06:34:14,687 INFO
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator - Marking
> the coordinator testing-gce-f67551b8-8867-41c5-b36e-4196d0f11e30:9092 (id:
> 2147483647 rack: null) dead for group myconsumer
> 2017-09-28 06:34:14,792 INFO
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator - Discovered
> coordinator testing-gce-f67551b8-8867-41c5-b36e-4196d0f11e30:9092 (id:
> 2147483647 rack: null) for group myconsumer.
> 2017-09-28 06:34:15,068 INFO
> org.apache.flink.runtime.state.DefaultOperatorStateBackend-
> DefaultOperatorStateBackend snapshot (In-Memory Stream Factory, synchronous
> part) in thread Thread[Async calls on Source: Custom Source -> Map -> Sink:
> Unnamed (1/1),5,Flink Task Threads] took 948 ms.
> 2017-09-28 06:34:15,164 WARN
> org.apache.flink.streaming.connectors.kafka.internal.Kafka09Fetcher -
> Committing offsets to Kafka failed. This does not compromise Flink's
> checkpoints.
> java.lang.IllegalArgumentException: Invalid offset: -915623761772
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.sendOffsetCommitRequest(ConsumerCoordinator.java:687)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.doCommitOffsetsAsync(ConsumerCoordinator.java:531)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsAsync(ConsumerCoordinator.java:499)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.commitAsync(KafkaConsumer.java:1181)
> at
> org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.run(KafkaConsumerThread.java:217)
> 2017-09-28 06:34:15,171 ERROR
> org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase - Async
> Kafka commit failed.
> java.lang.IllegalArgumentException: Invalid offset: -915623761772
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.sendOffsetCommitRequest(ConsumerCoordinator.java:687)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.doCommitOffsetsAsync(ConsumerCoordinator.java:531)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsAsync(ConsumerCoordinator.java:499)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.commitAsync(KafkaConsumer.java:1181)
> at
> org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.run(KafkaConsumerThread.java:217)
> {code}
> https://travis-ci.org/apache/flink/jobs/280722829
> [~pnowojski] did a first analysis that revealed this:
> In
> org/apache/flink/streaming/connectors/kafka/internal/Kafka09Fetcher.java:229
> this is being sent:
> {{long offsetToCommit
