dataartisans/yahoo-streaming-benchmark

An extension of Yahoo's Benchmarks

yahoo-streaming-benchmark's Issues

Compilation failing

Hello Team Flink,

I am trying to compile Flink app but it is failing, I have attached the log, Can you please help?

I was able to build the Flink app at https://github.com/yahoo/streaming-benchmarks, but not this repository.

Thanks
Sandesh

ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/AdvertisingTopologyFlinkWindows.java:[178,72] interface expected here
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/AdvertisingTopologyFlinkWindows.java:[181,87] cannot find symbol
[ERROR] symbol: class TriggerContext
[ERROR] location: class flink.benchmark.AdvertisingTopologyFlinkWindows.EventAndProcessingTimeTrigger
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/AdvertisingTopologyFlinkWindows.java:[181,12] cannot find symbol
[ERROR] symbol: class TriggerResult
[ERROR] location: class flink.benchmark.AdvertisingTopologyFlinkWindows.EventAndProcessingTimeTrigger
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/AdvertisingTopologyFlinkWindows.java:[193,68] cannot find symbol
[ERROR] symbol: class TriggerContext
[ERROR] location: class flink.benchmark.AdvertisingTopologyFlinkWindows.EventAndProcessingTimeTrigger
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/AdvertisingTopologyFlinkWindows.java:[193,12] cannot find symbol
[ERROR] symbol: class TriggerResult
[ERROR] location: class flink.benchmark.AdvertisingTopologyFlinkWindows.EventAndProcessingTimeTrigger
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/AdvertisingTopologyFlinkWindows.java:[198,73] cannot find symbol
[ERROR] symbol: class TriggerContext
[ERROR] location: class flink.benchmark.AdvertisingTopologyFlinkWindows.EventAndProcessingTimeTrigger
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/AdvertisingTopologyFlinkWindows.java:[198,12] cannot find symbol
[ERROR] symbol: class TriggerResult
[ERROR] location: class flink.benchmark.AdvertisingTopologyFlinkWindows.EventAndProcessingTimeTrigger
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/state/QueryableWindowOperatorEvicting.java:[36,38] cannot find symbol
[ERROR] symbol: class StateBackend
[ERROR] location: package org.apache.flink.runtime.state
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/state/QueryableWindowOperatorEvicting.java:[383,25] cannot find symbol
[ERROR] symbol: class StateBackend
[ERROR] location: class flink.benchmark.state.QueryableWindowOperatorEvicting.DataInputViewAsynchronousStateHandle
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/state/QueryableWindowOperatorEvicting.java:[389,33] cannot find symbol
[ERROR] symbol: class StateBackend
[ERROR] location: class flink.benchmark.state.QueryableWindowOperatorEvicting.DataInputViewAsynchronousStateHandle
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/state/QueryableWindowOperator.java:[31,38] cannot find symbol
[ERROR] symbol: class StateBackend
[ERROR] location: package org.apache.flink.runtime.state
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/state/QueryableWindowOperator.java:[338,13] cannot find symbol
[ERROR] symbol: class StateBackend
[ERROR] location: class flink.benchmark.state.QueryableWindowOperator.DataInputViewAsynchronousStateHandle
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/state/QueryableWindowOperator.java:[344,49] cannot find symbol
[ERROR] symbol: class StateBackend
[ERROR] location: class flink.benchmark.state.QueryableWindowOperator.DataInputViewAsynchronousStateHandle
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/AdvertisingTopologyFlinkWindows.java:[78,17] method trigger in class org.apache.flink.streaming.api.datastream.WindowedStream<T,K,W> cannot be applied to given types;
[ERROR] required: org.apache.flink.streaming.api.windowing.triggers.Trigger<? super org.apache.flink.api.java.tuple.Tuple3<java.lang.String,java.lang.String,java.lang.Long>,? super org.apache.flink.streaming.api.windowing.windows.TimeWindow>
[ERROR] found: flink.benchmark.AdvertisingTopologyFlinkWindows.EventAndProcessingTimeTrigger
[ERROR] reason: actual argument flink.benchmark.AdvertisingTopologyFlinkWindows.EventAndProcessingTimeTrigger cannot be converted to org.apache.flink.streaming.api.windowing.triggers.Trigger<? super org.apache.flink.api.java.tuple.Tuple3<java.lang.String,java.lang.String,java.lang.Long>,? super org.apache.flink.streaming.api.windowing.windows.TimeWindow> by method invocation conversion
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/AdvertisingTopologyFlinkWindows.java:[189,14] cannot find symbol
[ERROR] symbol: variable TriggerResult
[ERROR] location: class flink.benchmark.AdvertisingTopologyFlinkWindows.EventAndProcessingTimeTrigger
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/AdvertisingTopologyFlinkWindows.java:[180,5] method does not override or implement a method from a supertype
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/AdvertisingTopologyFlinkWindows.java:[194,14] cannot find symbol
[ERROR] symbol: variable TriggerResult
[ERROR] location: class flink.benchmark.AdvertisingTopologyFlinkWindows.EventAndProcessingTimeTrigger
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/AdvertisingTopologyFlinkWindows.java:[192,5] method does not override or implement a method from a supertype
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/AdvertisingTopologyFlinkWindows.java:[201,14] cannot find symbol
[ERROR] symbol: variable TriggerResult
[ERROR] location: class flink.benchmark.AdvertisingTopologyFlinkWindows.EventAndProcessingTimeTrigger
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/AdvertisingTopologyFlinkWindows.java:[197,5] method does not override or implement a method from a supertype
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/state/AdvertisingTopologyFlinkStateHighKeyCard.java:[63,7] method assignTimestamps in class org.apache.flink.streaming.api.datastream.DataStream cannot be applied to given types;
[ERROR] required: org.apache.flink.streaming.api.functions.TimestampExtractor<org.apache.flink.api.java.tuple.Tuple7<java.lang.String,java.lang.String,java.lang.String,java.lang.String,java.lang.String,java.lang.String,java.lang.String>>
[ERROR] found: flink.benchmark.state.AdvertisingTopologyFlinkStateHighKeyCard.AdTimestampExtractor
[ERROR] reason: actual argument flink.benchmark.state.AdvertisingTopologyFlinkStateHighKeyCard.AdTimestampExtractor cannot be converted to org.apache.flink.streaming.api.functions.TimestampExtractor<org.apache.flink.api.java.tuple.Tuple7<java.lang.String,java.lang.String,java.lang.String,java.lang.String,java.lang.String,java.lang.String,java.lang.String>> by method invocation conversion
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/state/AdvertisingTopologyFlinkStateHighKeyCard.java:[197,17] flink.benchmark.state.AdvertisingTopologyFlinkStateHighKeyCard.AdTimestampExtractor is not abstract and does not override abstract method extractAscendingTimestamp(org.apache.flink.api.java.tuple.Tuple7<java.lang.String,java.lang.String,java.lang.String,java.lang.String,java.lang.String,java.lang.String,java.lang.String>) in org.apache.flink.streaming.api.functions.AscendingTimestampExtractor
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/state/AdvertisingTopologyFlinkStateHighKeyCard.java:[198,5] method does not override or implement a method from a supertype
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/state/QueryableWindowOperatorEvicting.java:[399,37] package StateBackend does not exist
[ERROR] /home/sandesh/dev/yahoo-streaming-benchmark/flink-benchmarks/src/main/java/flink/benchmark/state/QueryableWindowOperator.java:[353,19] package StateBackend does not exist
[ERROR] -> [Help 1]
[ERROR]
[ERROR] To see the full stack trace of the errors, re-run Maven with the -e switch.
[ERROR] Re-run Maven using the -X switch to enable full debug logging.
[ERROR]
[ERROR] For more information about the errors and possible solutions, please read the following articles:
[ERROR] [Help 1] http://cwiki.apache.org/confluence/display/MAVEN/MojoFailureException
[ERROR]
[ERROR] After correcting the problems, you can resume the build with the command
[ERROR] mvn -rf :flink-benchmarks

AdBlock — best ad blocker - Chrome Web Store

https://chrome.google.com/webstore/detail/adblock-%E2%80%94-best-ad-blocker/gighmmpiobklfepjocnamgkkbiglidom?hl=en-US&gl=001

How to make sense of update.txt tile

I'm trying to benchmark Storm and cannot figure out how to calculate the latency. The document says to calculate latency with the formula

window.final_event_latency = (window.last_updated_at – window.timestamp) – window.duration

I believe (window.last_updated_at – window.timestamp) this is the value in update.txt file. For window.duration of 10 seconds, I get values below 10 seconds in the update.txt file essentially making latency negative.

Flink job execution failed

Hi dataArtisans,

I'm trying to run the flink benchmark (AdvertisingTopologyNative). I manually started Zookeeper, Kafka, Redis and Flink job manager using:

./stream-bench.sh START_ZK
./stream-bench.sh START_KAFKA
./stream-bench.sh START_REDIS
./stream-bench.sh START_FLINK

Then I started the flink job using ./stream-bench.sh START_FLINK_PROCESSING and after some time the job switches to Failed. Could you please help?

Here is the full output:

ubuntu@my-bench:~/yahoo-streaming-benchmark$ ./stream-bench.sh START_FLINK_PROCESSING
[2016-05-23 10:40:58,041] INFO Closing socket connection to /127.0.0.1. (kafka.network.Processor)
ubuntu@my-bench:~/yahoo-streaming-benchmark$ 05/23/2016 10:40:59        Job execution switched to status RUNNING.
05/23/2016 10:40:59     Source: Custom Source -> Flat Map(1/1) switched to SCHEDULED
05/23/2016 10:40:59     Source: Custom Source -> Flat Map(1/1) switched to DEPLOYING
05/23/2016 10:40:59     Flat Map -> Filter -> Projection -> Flat Map(1/1) switched to SCHEDULED
05/23/2016 10:40:59     Flat Map -> Filter -> Projection -> Flat Map(1/1) switched to DEPLOYING
05/23/2016 10:40:59     Flat Map(1/1) switched to SCHEDULED
05/23/2016 10:40:59     Flat Map(1/1) switched to DEPLOYING
05/23/2016 10:41:00     Source: Custom Source -> Flat Map(1/1) switched to RUNNING
05/23/2016 10:41:00     Flat Map(1/1) switched to RUNNING
05/23/2016 10:41:00     Flat Map -> Filter -> Projection -> Flat Map(1/1) switched to RUNNING
05/23/2016 10:42:01     Flat Map -> Filter -> Projection -> Flat Map(1/1) switched to CANCELED
05/23/2016 10:42:01     Source: Custom Source -> Flat Map(1/1) switched to FAILED
java.lang.NumberFormatException: For input string: "2181null"
        at java.lang.NumberFormatException.forInputString(NumberFormatException.java:65)
        at java.lang.Integer.parseInt(Integer.java:580)
        at java.lang.Integer.parseInt(Integer.java:615)
        at org.apache.zookeeper.client.ConnectStringParser.<init>(ConnectStringParser.java:72)
        at org.apache.zookeeper.ZooKeeper.<init>(ZooKeeper.java:443)
        at org.apache.flink.shaded.org.apache.curator.utils.DefaultZookeeperFactory.newZooKeeper(DefaultZookeeperFactory.java:29)
        at org.apache.flink.shaded.org.apache.curator.framework.imps.CuratorFrameworkImpl$2.newZooKeeper(CuratorFrameworkImpl.java:146)
        at org.apache.flink.shaded.org.apache.curator.HandleHolder$1.getZooKeeper(HandleHolder.java:94)
        at org.apache.flink.shaded.org.apache.curator.HandleHolder.internalClose(HandleHolder.java:128)
        at org.apache.flink.shaded.org.apache.curator.HandleHolder.closeAndReset(HandleHolder.java:77)
        at org.apache.flink.shaded.org.apache.curator.ConnectionState.reset(ConnectionState.java:217)
        at org.apache.flink.shaded.org.apache.curator.ConnectionState.checkTimeouts(ConnectionState.java:193)
        at org.apache.flink.shaded.org.apache.curator.ConnectionState.getZooKeeper(ConnectionState.java:87)
        at org.apache.flink.shaded.org.apache.curator.CuratorZookeeperClient.getZooKeeper(CuratorZookeeperClient.java:115)
        at org.apache.flink.shaded.org.apache.curator.utils.EnsurePath$InitialHelper$1.call(EnsurePath.java:148)
        at org.apache.flink.shaded.org.apache.curator.RetryLoop.callWithRetry(RetryLoop.java:107)
        at org.apache.flink.shaded.org.apache.curator.utils.EnsurePath$InitialHelper.ensure(EnsurePath.java:141)
        at org.apache.flink.shaded.org.apache.curator.utils.EnsurePath.ensure(EnsurePath.java:99)
        at org.apache.flink.streaming.connectors.kafka.internals.ZookeeperOffsetHandler.getOffsetFromZooKeeper(ZookeeperOffsetHandler.java:126)
        at org.apache.flink.streaming.connectors.kafka.internals.ZookeeperOffsetHandler.getOffsets(ZookeeperOffsetHandler.java:95)
        at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08.open(FlinkKafkaConsumer08.java:284)
        at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:38)
        at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:91)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:317)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:215)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559)
        at java.lang.Thread.run(Thread.java:745)

05/23/2016 10:42:01     Job execution switched to status FAILING.
java.lang.NumberFormatException: For input string: "2181null"
        at java.lang.NumberFormatException.forInputString(NumberFormatException.java:65)
        at java.lang.Integer.parseInt(Integer.java:580)
        at java.lang.Integer.parseInt(Integer.java:615)
        at org.apache.zookeeper.client.ConnectStringParser.<init>(ConnectStringParser.java:72)
        at org.apache.zookeeper.ZooKeeper.<init>(ZooKeeper.java:443)
        at org.apache.flink.shaded.org.apache.curator.utils.DefaultZookeeperFactory.newZooKeeper(DefaultZookeeperFactory.java:29)
        at org.apache.flink.shaded.org.apache.curator.framework.imps.CuratorFrameworkImpl$2.newZooKeeper(CuratorFrameworkImpl.java:146)
        at org.apache.flink.shaded.org.apache.curator.HandleHolder$1.getZooKeeper(HandleHolder.java:94)
        at org.apache.flink.shaded.org.apache.curator.HandleHolder.internalClose(HandleHolder.java:128)
        at org.apache.flink.shaded.org.apache.curator.HandleHolder.closeAndReset(HandleHolder.java:77)
        at org.apache.flink.shaded.org.apache.curator.ConnectionState.reset(ConnectionState.java:217)
        at org.apache.flink.shaded.org.apache.curator.ConnectionState.checkTimeouts(ConnectionState.java:193)
        at org.apache.flink.shaded.org.apache.curator.ConnectionState.getZooKeeper(ConnectionState.java:87)
        at org.apache.flink.shaded.org.apache.curator.CuratorZookeeperClient.getZooKeeper(CuratorZookeeperClient.java:115)
        at org.apache.flink.shaded.org.apache.curator.utils.EnsurePath$InitialHelper$1.call(EnsurePath.java:148)
        at org.apache.flink.shaded.org.apache.curator.RetryLoop.callWithRetry(RetryLoop.java:107)
        at org.apache.flink.shaded.org.apache.curator.utils.EnsurePath$InitialHelper.ensure(EnsurePath.java:141)
        at org.apache.flink.shaded.org.apache.curator.utils.EnsurePath.ensure(EnsurePath.java:99)
        at org.apache.flink.streaming.connectors.kafka.internals.ZookeeperOffsetHandler.getOffsetFromZooKeeper(ZookeeperOffsetHandler.java:126)
        at org.apache.flink.streaming.connectors.kafka.internals.ZookeeperOffsetHandler.getOffsets(ZookeeperOffsetHandler.java:95)
        at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08.open(FlinkKafkaConsumer08.java:284)
        at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:38)
        at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:91)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:317)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:215)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559)
        at java.lang.Thread.run(Thread.java:745)
05/23/2016 10:42:01     Flat Map(1/1) switched to CANCELING
05/23/2016 10:42:01     Flat Map(1/1) switched to CANCELED
05/23/2016 10:42:01     Job execution switched to status FAILED.

------------------------------------------------------------
 The program finished with the following exception:

org.apache.flink.client.program.ProgramInvocationException: The program execution failed: Job execution failed.
        at org.apache.flink.client.program.Client.runBlocking(Client.java:381)
        at org.apache.flink.client.program.Client.runBlocking(Client.java:355)
        at org.apache.flink.streaming.api.environment.StreamContextEnvironment.execute(StreamContextEnvironment.java:65)
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1170)
        at flink.benchmark.AdvertisingTopologyNative.main(AdvertisingTopologyNative.java:69)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:505)
        at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:403)
        at org.apache.flink.client.program.Client.runBlocking(Client.java:248)
        at org.apache.flink.client.CliFrontend.executeProgramBlocking(CliFrontend.java:866)
        at org.apache.flink.client.CliFrontend.run(CliFrontend.java:333)
        at org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:1189)
        at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1239)
Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
        at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply$mcV$sp(JobManager.scala:716)
        at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply(JobManager.scala:662)
        at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply(JobManager.scala:662)
        at scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)
        at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)
        at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:41)
        at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:401)
        at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
        at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.pollAndExecAll(ForkJoinPool.java:1253)
        at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1346)
        at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
        at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
Caused by: java.lang.NumberFormatException: For input string: "2181null"
        at java.lang.NumberFormatException.forInputString(NumberFormatException.java:65)
        at java.lang.Integer.parseInt(Integer.java:580)
        at java.lang.Integer.parseInt(Integer.java:615)
        at org.apache.zookeeper.client.ConnectStringParser.<init>(ConnectStringParser.java:72)
        at org.apache.zookeeper.ZooKeeper.<init>(ZooKeeper.java:443)
        at org.apache.flink.shaded.org.apache.curator.utils.DefaultZookeeperFactory.newZooKeeper(DefaultZookeeperFactory.java:29)
        at org.apache.flink.shaded.org.apache.curator.framework.imps.CuratorFrameworkImpl$2.newZooKeeper(CuratorFrameworkImpl.java:146)
        at org.apache.flink.shaded.org.apache.curator.HandleHolder$1.getZooKeeper(HandleHolder.java:94)
        at org.apache.flink.shaded.org.apache.curator.HandleHolder.internalClose(HandleHolder.java:128)
        at org.apache.flink.shaded.org.apache.curator.HandleHolder.closeAndReset(HandleHolder.java:77)
        at org.apache.flink.shaded.org.apache.curator.ConnectionState.reset(ConnectionState.java:217)
        at org.apache.flink.shaded.org.apache.curator.ConnectionState.checkTimeouts(ConnectionState.java:193)
        at org.apache.flink.shaded.org.apache.curator.ConnectionState.getZooKeeper(ConnectionState.java:87)
        at org.apache.flink.shaded.org.apache.curator.CuratorZookeeperClient.getZooKeeper(CuratorZookeeperClient.java:115)
        at org.apache.flink.shaded.org.apache.curator.utils.EnsurePath$InitialHelper$1.call(EnsurePath.java:148)
        at org.apache.flink.shaded.org.apache.curator.RetryLoop.callWithRetry(RetryLoop.java:107)
        at org.apache.flink.shaded.org.apache.curator.utils.EnsurePath$InitialHelper.ensure(EnsurePath.java:141)
        at org.apache.flink.shaded.org.apache.curator.utils.EnsurePath.ensure(EnsurePath.java:99)
        at org.apache.flink.streaming.connectors.kafka.internals.ZookeeperOffsetHandler.getOffsetFromZooKeeper(ZookeeperOffsetHandler.java:126)
        at org.apache.flink.streaming.connectors.kafka.internals.ZookeeperOffsetHandler.getOffsets(ZookeeperOffsetHandler.java:95)
        at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08.open(FlinkKafkaConsumer08.java:284)
        at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:38)
        at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:91)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:317)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:215)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559)
        at java.lang.Thread.run(Thread.java:745)
18401:M 23 May 11:43:06.038 * 100 changes in 300 seconds. Saving...
18401:M 23 May 11:43:06.040 * Background saving started by pid 18982
18982:C 23 May 11:43:06.089 * DB saved on disk
18982:C 23 May 11:43:06.090 * RDB: 6 MB of memory used by copy-on-write
18401:M 23 May 11:43:06.140 * Background saving terminated with success

Thank you.

I'm Glad My Mom Died

I'm Glad My Mom Died https://open.spotify.com/show/7sn0W2eNkETlJgbgmhijDI