?????? Flink Cli ????????

Z-Z Sun, 19 Jul 2020 20:34:01 -0700

????taskmanager????????????????????????????????cli????????????webui??????????????
2020-07-20 03:29:25,959 WARN&nbsp; 
org.apache.kafka.clients.consumer.ConsumerConfig&nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; - The configuration 'value.serializer' was supplied but 
isn't a known config.
2020-07-20 03:29:25,959 INFO&nbsp; 
org.apache.kafka.common.utils.AppInfoParser&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp;- Kafka version : 0.11.0.2
2020-07-20 03:29:25,959 INFO&nbsp; 
org.apache.kafka.common.utils.AppInfoParser&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp;- Kafka commitId : 73be1e1168f91ee2
2020-07-20 03:29:25,974 ERROR 
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder&nbsp; 
- Caught unexpected exception.
java.lang.ArrayIndexOutOfBoundsException: 0
        at 
org.apache.flink.contrib.streaming.state.snapshot.RocksSnapshotUtil.hasMetaDataFollowsFlag(RocksSnapshotUtil.java:45)
        at 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restoreKVStateData(RocksDBFullRestoreOperation.java:223)
        at 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restoreKeyGroupsInStateHandle(RocksDBFullRestoreOperation.java:168)
        at 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restore(RocksDBFullRestoreOperation.java:151)
        at 
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:279)
        at 
org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:548)
        at 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:288)
        at 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:142)
        at 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:121)
        at 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:304)
        at 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:131)
        at 
org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:255)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTask.initializeStateAndOpen(StreamTask.java:1006)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$0(StreamTask.java:454)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:94)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:449)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:461)
        at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:707)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:532)
        at java.lang.Thread.run(Thread.java:748)
2020-07-20 03:29:25,974 WARN&nbsp; 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure&nbsp; - 
Exception while restoring keyed state backend for 
StreamMap_caf773fe289bfdb867e0b4bd0c431c5f_(1/1) from alternative (1/1), will 
retry while more alternatives are available.
org.apache.flink.runtime.state.BackendBuildingException: Caught unexpected 
exception.
        at 
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:336)
        at 
org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:548)
        at 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:288)
        at 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:142)
        at 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:121)
        at 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:304)
        at 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:131)
        at 
org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:255)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTask.initializeStateAndOpen(StreamTask.java:1006)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$0(StreamTask.java:454)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:94)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:449)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:461)
        at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:707)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:532)
        at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.ArrayIndexOutOfBoundsException: 0
        at 
org.apache.flink.contrib.streaming.state.snapshot.RocksSnapshotUtil.hasMetaDataFollowsFlag(RocksSnapshotUtil.java:45)
        at 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restoreKVStateData(RocksDBFullRestoreOperation.java:223)
        at 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restoreKeyGroupsInStateHandle(RocksDBFullRestoreOperation.java:168)
        at 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restore(RocksDBFullRestoreOperation.java:151)
        at 
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:279)
        ... 15 more
2020-07-20 03:29:25,975 INFO&nbsp; 
org.apache.kafka.clients.producer.KafkaProducer&nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp;- Closing the Kafka producer with timeoutMillis = 
9223372036854775807 ms.
2020-07-20 03:29:25,979 INFO&nbsp; 
org.apache.flink.runtime.taskmanager.Task&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;- Map -&gt; Filter -&gt; Sink: Unnamed 
(1/1) (ed554502aa995fe53f1cf0cb8adf633c) switched from RUNNING to FAILED.
java.lang.Exception: Exception while creating StreamOperatorStateContext.
        at 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:191)
        at 
org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:255)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTask.initializeStateAndOpen(StreamTask.java:1006)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$0(StreamTask.java:454)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:94)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:449)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:461)
        at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:707)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:532)
        at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.util.FlinkException: Could not restore keyed state 
backend for StreamMap_caf773fe289bfdb867e0b4bd0c431c5f_(1/1) from any of the 1 
provided restore options.
        at 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)
        at 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:304)
        at 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:131)
        ... 9 more
Caused by: org.apache.flink.runtime.state.BackendBuildingException: Caught 
unexpected exception.
        at 
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:336)
        at 
org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:548)
        at 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:288)
        at 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:142)
        at 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:121)
        ... 11 more
Caused by: java.lang.ArrayIndexOutOfBoundsException: 0
        at 
org.apache.flink.contrib.streaming.state.snapshot.RocksSnapshotUtil.hasMetaDataFollowsFlag(RocksSnapshotUtil.java:45)
        at 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restoreKVStateData(RocksDBFullRestoreOperation.java:223)
        at 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restoreKeyGroupsInStateHandle(RocksDBFullRestoreOperation.java:168)
        at 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restore(RocksDBFullRestoreOperation.java:151)
        at 
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:279)








??????????
??????savepoint????????RocksDBStateBackend????????
????savepoint??????webui ??????????????????????????IDE??????savepoint??




------------------ ???????? ------------------
??????:                                                                         
                                               "user-zh"                        
                                                            
<[email protected]&gt;;
????????:&nbsp;2020??7??19??(??????) ????8:22
??????:&nbsp;"user-zh"<[email protected]&gt;;

????:&nbsp;Re: Flink Cli ????????



Hi

?????????????????????????????????????? EOF ????????????????
1 ???????? savepoint ?????? RocksDBStateBackend ????????
2 ???????????? DFS ???? savepoint ???????????????????? DFS 
??????????????????????????????????

Best,
Congxian


Z-Z <[email protected]&gt; ??2020??7??17?????? ????11:10??????

&gt; Flink 1.10.0 ,taskmanager??????????????
&gt;
&gt;
&gt; 2020-07-17 15:06:43,913 ERROR
&gt; 
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder&amp;nbsp;
&gt; - Caught unexpected exception.
&gt; java.io.EOFException
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at 
java.io.DataInputStream.readFully(DataInputStream.java:197)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at 
java.io.DataInputStream.readFully(DataInputStream.java:169)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer.deserialize(BytePrimitiveArraySerializer.java:85)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restoreKVStateData(RocksDBFullRestoreOperation.java:221)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restoreKeyGroupsInStateHandle(RocksDBFullRestoreOperation.java:168)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restore(RocksDBFullRestoreOperation.java:151)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:279)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:548)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:288)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:142)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:121)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:304)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:131)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:255)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTask.initializeStateAndOpen(StreamTask.java:1006)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$0(StreamTask.java:454)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:94)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:449)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:461)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at 
org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:707)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at 
org.apache.flink.runtime.taskmanager.Task.run(Task.java:532)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at 
java.lang.Thread.run(Thread.java:748)
&gt; 2020-07-17 15:06:43,914 WARN&amp;nbsp;
&gt; 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure&amp;nbsp; -
&gt; Exception while restoring keyed state backend for
&gt; KeyedCoProcessOperator_00360b8021b192d84949201d4fea80f2_(1/1) from
&gt; alternative (1/1), will retry while more alternatives are available.
&gt; org.apache.flink.runtime.state.BackendBuildingException: Caught unexpected
&gt; exception.
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:336)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:548)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:288)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:142)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:121)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:304)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:131)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:255)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTask.initializeStateAndOpen(StreamTask.java:1006)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$0(StreamTask.java:454)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:94)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:449)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:461)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at 
org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:707)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at 
org.apache.flink.runtime.taskmanager.Task.run(Task.java:532)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at 
java.lang.Thread.run(Thread.java:748)
&gt; Caused by: java.io.EOFException
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at 
java.io.DataInputStream.readFully(DataInputStream.java:197)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at 
java.io.DataInputStream.readFully(DataInputStream.java:169)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer.deserialize(BytePrimitiveArraySerializer.java:85)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restoreKVStateData(RocksDBFullRestoreOperation.java:221)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restoreKeyGroupsInStateHandle(RocksDBFullRestoreOperation.java:168)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restore(RocksDBFullRestoreOperation.java:151)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:279)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; ... 15 more
&gt; 2020-07-17 15:06:43,915 INFO&amp;nbsp;
&gt; org.apache.kafka.clients.producer.KafkaProducer&amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- Closing the Kafka producer 
with timeoutMillis
&gt; = 9223372036854775807 ms.
&gt; 2020-07-17 15:06:43,918 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskmanager.Task&amp;nbsp; &amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- 
Co-Keyed-Process -&amp;gt; Flat Map
&gt; -&amp;gt; Sink: Unnamed (1/1) (bb8f0a84e07ef90b1e11ca2825e0efab) switched 
from
&gt; RUNNING to FAILED.
&gt; java.lang.Exception: Exception while creating StreamOperatorStateContext.
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:191)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:255)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTask.initializeStateAndOpen(StreamTask.java:1006)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$0(StreamTask.java:454)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:94)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:449)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:461)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at 
org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:707)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at 
org.apache.flink.runtime.taskmanager.Task.run(Task.java:532)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at 
java.lang.Thread.run(Thread.java:748)
&gt; Caused by: org.apache.flink.util.FlinkException: Could not restore keyed
&gt; state backend for
&gt; KeyedCoProcessOperator_00360b8021b192d84949201d4fea80f2_(1/1) from any of
&gt; the 1 provided restore options.
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:304)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:131)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; ... 9 more
&gt; Caused by: org.apache.flink.runtime.state.BackendBuildingException: Caught
&gt; unexpected exception.
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:336)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:548)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:288)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:142)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:121)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; ... 11 more
&gt; Caused by: java.io.EOFException
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at 
java.io.DataInputStream.readFully(DataInputStream.java:197)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at 
java.io.DataInputStream.readFully(DataInputStream.java:169)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer.deserialize(BytePrimitiveArraySerializer.java:85)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restoreKVStateData(RocksDBFullRestoreOperation.java:221)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restoreKeyGroupsInStateHandle(RocksDBFullRestoreOperation.java:168)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restore(RocksDBFullRestoreOperation.java:151)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:279)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; ... 15 more
&gt; 2020-07-17 15:06:43,919 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskmanager.Task&amp;nbsp; &amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- Freeing 
task resources for
&gt; Co-Keyed-Process -&amp;gt; Flat Map -&amp;gt; Sink: Unnamed (1/1)
&gt; (bb8f0a84e07ef90b1e11ca2825e0efab).
&gt; 2020-07-17 15:06:43,919 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskmanager.Task&amp;nbsp; &amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- 
Ensuring all FileSystem streams
&gt; are closed for task Co-Keyed-Process -&amp;gt; Flat Map -&amp;gt; Sink: 
Unnamed
&gt; (1/1) (bb8f0a84e07ef90b1e11ca2825e0efab) [FAILED]
&gt; 2020-07-17 15:06:43,931 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskexecutor.TaskExecutor&amp;nbsp; &amp;nbsp; 
&amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; - Un-registering task and sending final 
execution
&gt; state FAILED to JobManager for task Co-Keyed-Process -&amp;gt; Flat Map 
-&amp;gt;
&gt; Sink: Unnamed (1/1) bb8f0a84e07ef90b1e11ca2825e0efab.
&gt; 2020-07-17 15:06:43,947 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskmanager.Task&amp;nbsp; &amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- 
Attempting to cancel task
&gt; Source: Custom Source -&amp;gt; Flat Map (1/1)
&gt; (9cb8dcd4982223adcb6f007f1ffccdce).
&gt; 2020-07-17 15:06:43,947 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskmanager.Task&amp;nbsp; &amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- Source: 
Custom Source -&amp;gt; Flat
&gt; Map (1/1) (9cb8dcd4982223adcb6f007f1ffccdce) switched from RUNNING to
&gt; CANCELING.
&gt; 2020-07-17 15:06:43,947 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskmanager.Task&amp;nbsp; &amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- 
Triggering cancellation of task
&gt; code Source: Custom Source -&amp;gt; Flat Map (1/1)
&gt; (9cb8dcd4982223adcb6f007f1ffccdce).
&gt; 2020-07-17 15:06:43,949 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskmanager.Task&amp;nbsp; &amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- 
Attempting to cancel task
&gt; Source: Custom Source (1/1) (00621ff5d788d00c73ccaaea04717600).
&gt; 2020-07-17 15:06:43,949 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskmanager.Task&amp;nbsp; &amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- Source: 
Custom Source (1/1)
&gt; (00621ff5d788d00c73ccaaea04717600) switched from RUNNING to CANCELING.
&gt; 2020-07-17 15:06:43,949 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskmanager.Task&amp;nbsp; &amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- 
Triggering cancellation of task
&gt; code Source: Custom Source (1/1) (00621ff5d788d00c73ccaaea04717600).
&gt; 2020-07-17 15:06:43,954 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskmanager.Task&amp;nbsp; &amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- Source: 
Custom Source -&amp;gt; Flat
&gt; Map (1/1) (9cb8dcd4982223adcb6f007f1ffccdce) switched from CANCELING to
&gt; CANCELED.
&gt; 2020-07-17 15:06:43,954 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskmanager.Task&amp;nbsp; &amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- Freeing 
task resources for
&gt; Source: Custom Source -&amp;gt; Flat Map (1/1)
&gt; (9cb8dcd4982223adcb6f007f1ffccdce).
&gt; 2020-07-17 15:06:43,954 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskmanager.Task&amp;nbsp; &amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- 
Ensuring all FileSystem streams
&gt; are closed for task Source: Custom Source -&amp;gt; Flat Map (1/1)
&gt; (9cb8dcd4982223adcb6f007f1ffccdce) [CANCELED]
&gt; 2020-07-17 15:06:43,954 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskmanager.Task&amp;nbsp; &amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- Source: 
Custom Source (1/1)
&gt; (00621ff5d788d00c73ccaaea04717600) switched from CANCELING to CANCELED.
&gt; 2020-07-17 15:06:43,955 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskmanager.Task&amp;nbsp; &amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- Freeing 
task resources for
&gt; Source: Custom Source (1/1) (00621ff5d788d00c73ccaaea04717600).
&gt; 2020-07-17 15:06:43,954 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskexecutor.TaskExecutor&amp;nbsp; &amp;nbsp; 
&amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; - Un-registering task and sending final 
execution
&gt; state CANCELED to JobManager for task Source: Custom Source -&amp;gt; Flat 
Map
&gt; (1/1) 9cb8dcd4982223adcb6f007f1ffccdce.
&gt; 2020-07-17 15:06:43,962 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskmanager.Task&amp;nbsp; &amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- 
Ensuring all FileSystem streams
&gt; are closed for task Source: Custom Source (1/1)
&gt; (00621ff5d788d00c73ccaaea04717600) [CANCELED]
&gt; 2020-07-17 15:06:43,962 INFO&amp;nbsp;
&gt; org.apache.flink.runtime.taskexecutor.TaskExecutor&amp;nbsp; &amp;nbsp; 
&amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; - Un-registering task and sending final 
execution
&gt; state CANCELED to JobManager for task Source: Custom Source (1/1)
&gt; 00621ff5d788d00c73ccaaea04717600.
&gt; 2020-07-17 15:06:44,077 WARN&amp;nbsp;
&gt; org.apache.kafka.clients.consumer.ConsumerConfig&amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; - The configuration 
'transaction.timeout.ms' was
&gt; supplied but isn't a known config.
&gt; 2020-07-17 15:06:44,077 WARN&amp;nbsp;
&gt; org.apache.kafka.clients.consumer.ConsumerConfig&amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; - The configuration 'key.serializer' was 
supplied but
&gt; isn't a known config.
&gt; 2020-07-17 15:06:44,077 WARN&amp;nbsp;
&gt; org.apache.kafka.clients.consumer.ConsumerConfig&amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; - The configuration 'value.serializer' 
was supplied
&gt; but isn't a known config.
&gt; 2020-07-17 15:06:44,077 INFO&amp;nbsp;
&gt; org.apache.kafka.common.utils.AppInfoParser&amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- Kafka 
version : 0.11.0.2
&gt; 2020-07-17 15:06:44,077 INFO&amp;nbsp;
&gt; org.apache.kafka.common.utils.AppInfoParser&amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- Kafka 
commitId : 73be1e1168f91ee2
&gt; 2020-07-17 15:06:44,077 WARN&amp;nbsp;
&gt; org.apache.kafka.common.utils.AppInfoParser&amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- Error 
registering AppInfo mbean
&gt; javax.management.InstanceAlreadyExistsException:
&gt; kafka.consumer:type=app-info,id=consumer-3
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at 
com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:437)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1898)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:966)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:900)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:324)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.kafka.common.utils.AppInfoParser.registerAppInfo(AppInfoParser.java:58)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.kafka.clients.consumer.KafkaConsumer.<init&amp;gt;(KafkaConsumer.java:757)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.kafka.clients.consumer.KafkaConsumer.<init&amp;gt;(KafkaConsumer.java:633)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.kafka.clients.consumer.KafkaConsumer.<init&amp;gt;(KafkaConsumer.java:615)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.getConsumer(KafkaConsumerThread.java:502)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.run(KafkaConsumerThread.java:181)
&gt; 2020-07-17 15:06:44,079 INFO&amp;nbsp;
&gt; org.apache.kafka.common.utils.AppInfoParser&amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- Kafka 
version : 0.11.0.2
&gt; 2020-07-17 15:06:44,079 INFO&amp;nbsp;
&gt; org.apache.kafka.common.utils.AppInfoParser&amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- Kafka 
commitId : 73be1e1168f91ee2
&gt; 2020-07-17 15:06:44,079 WARN&amp;nbsp;
&gt; org.apache.kafka.common.utils.AppInfoParser&amp;nbsp; &amp;nbsp; 
&amp;nbsp; &amp;nbsp;
&gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp;- Error 
registering AppInfo mbean
&gt; javax.management.InstanceAlreadyExistsException:
&gt; kafka.consumer:type=app-info,id=consumer-4
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at 
com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:437)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1898)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:966)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:900)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:324)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.kafka.common.utils.AppInfoParser.registerAppInfo(AppInfoParser.java:58)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.kafka.clients.consumer.KafkaConsumer.<init&amp;gt;(KafkaConsumer.java:757)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.kafka.clients.consumer.KafkaConsumer.<init&amp;gt;(KafkaConsumer.java:633)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.kafka.clients.consumer.KafkaConsumer.<init&amp;gt;(KafkaConsumer.java:615)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.getConsumer(KafkaConsumerThread.java:502)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; at
&gt; 
org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.run(KafkaConsumerThread.java:181)
&gt;
&gt;
&gt;
&gt;
&gt;
&gt; ------------------&amp;nbsp;????????&amp;nbsp;------------------
&gt; ??????:
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 &nbsp; "user-zh"
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 &nbsp; <
&gt; [email protected]&amp;gt;;
&gt; ????????:&amp;nbsp;2020??7??17??(??????) ????10:52
&gt; ??????:&amp;nbsp;"user-zh"<[email protected]&amp;gt;;
&gt;
&gt; ????:&amp;nbsp;Re: Flink Cli ????????
&gt;
&gt;
&gt;
&gt; Hi
&gt;
&gt; ???????????????????? Flink ????????????????&amp;nbsp; Co-Process (1/1)
&gt; (d0309f26a545e74643382ed3f758269b) ???? tm ?? log 
?????????????????????????????? 083f69d029de
&gt; ????????????
&gt;
&gt; Best,
&gt; Congxian
&gt;
&gt;
&gt; Z-Z <[email protected]&amp;gt; ??2020??7??17?????? ????6:22??????
&gt;
&gt; &amp;gt;
&gt; 
????????????????????????????????????????????restAPI??????????????????????????????savepoint(??????/jobs/overview
&gt; &amp;gt; ---&amp;amp;gt; /jobs/{jobid}/savepoints ---&amp;amp;gt;
&gt; &amp;gt;
&gt; 
/jobs/{jobid}/savepoints/{triggerid})??????????flink????????savepoint????????????????????????webui????jar??????savepoint??????????????????????????
&gt; &amp;gt; 2020-07-17 09:51:48,925 INFO&amp;amp;nbsp;
&gt; &amp;gt;
&gt; 
org.apache.flink.runtime.resourcemanager.StandaloneResourceManager&amp;amp;nbsp;
&gt; -
&gt; &amp;gt; Request slot with profile ResourceProfile{UNKNOWN} for job
&gt; &amp;gt; 7639673873b707aa86c4387aa7b4aac3 with allocation id
&gt; &amp;gt; e8865cdbfe4c3c33099c7112bc2e3231.
&gt; &amp;gt; 2020-07-17 09:51:48,952 INFO&amp;amp;nbsp;
&gt; &amp;gt; 
org.apache.flink.runtime.executiongraph.ExecutionGraph&amp;amp;nbsp;
&gt; &amp;amp;nbsp; &amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;nbsp; - Source: Custom Source -&amp;amp;gt; Filter (1/1)
&gt; &amp;gt; (1177659bff014e8dbc3f0508055d4307) switched from SCHEDULED to
&gt; DEPLOYING.
&gt; &amp;gt; 2020-07-17 09:51:48,952 INFO&amp;amp;nbsp;
&gt; &amp;gt; 
org.apache.flink.runtime.executiongraph.ExecutionGraph&amp;amp;nbsp;
&gt; &amp;amp;nbsp; &amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;nbsp; - Deploying Source: Custom Source -&amp;amp;gt; 
Filter (1/1)
&gt; (attempt #0) to
&gt; &amp;gt; e63d829deafc144cd82efd73979dd056 @ 083f69d029de (dataPort=35758)
&gt; &amp;gt; 2020-07-17 09:51:48,953 INFO&amp;amp;nbsp;
&gt; &amp;gt; 
org.apache.flink.runtime.executiongraph.ExecutionGraph&amp;amp;nbsp;
&gt; &amp;amp;nbsp; &amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;nbsp; - Source: Custom Source (1/1)
&gt; (141f0dc22b624b39e21127f637ba63c2)
&gt; &amp;gt; switched from SCHEDULED to DEPLOYING.
&gt; &amp;gt; 2020-07-17 09:51:48,953 INFO&amp;amp;nbsp;
&gt; &amp;gt; 
org.apache.flink.runtime.executiongraph.ExecutionGraph&amp;amp;nbsp;
&gt; &amp;amp;nbsp; &amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;nbsp; - Deploying Source: Custom Source (1/1) (attempt 
#0) to
&gt; &amp;gt; e63d829deafc144cd82efd73979dd056 @ 083f69d029de (dataPort=35758)
&gt; &amp;gt; 2020-07-17 09:51:48,954 INFO&amp;amp;nbsp;
&gt; &amp;gt; 
org.apache.flink.runtime.executiongraph.ExecutionGraph&amp;amp;nbsp;
&gt; &amp;amp;nbsp; &amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;nbsp; - Source: Custom Source (1/1)
&gt; (274b3df03e1fab627059c1a78e4a26da)
&gt; &amp;gt; switched from SCHEDULED to DEPLOYING.
&gt; &amp;gt; 2020-07-17 09:51:48,954 INFO&amp;amp;nbsp;
&gt; &amp;gt; 
org.apache.flink.runtime.executiongraph.ExecutionGraph&amp;amp;nbsp;
&gt; &amp;amp;nbsp; &amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;nbsp; - Deploying Source: Custom Source (1/1) (attempt 
#0) to
&gt; &amp;gt; e63d829deafc144cd82efd73979dd056 @ 083f69d029de (dataPort=35758)
&gt; &amp;gt; 2020-07-17 09:51:48,954 INFO&amp;amp;nbsp;
&gt; &amp;gt; 
org.apache.flink.runtime.executiongraph.ExecutionGraph&amp;amp;nbsp;
&gt; &amp;amp;nbsp; &amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;nbsp; - Co-Process (1/1) 
(d0309f26a545e74643382ed3f758269b)
&gt; switched from
&gt; &amp;gt; SCHEDULED to DEPLOYING.
&gt; &amp;gt; 2020-07-17 09:51:48,954 INFO&amp;amp;nbsp;
&gt; &amp;gt; 
org.apache.flink.runtime.executiongraph.ExecutionGraph&amp;amp;nbsp;
&gt; &amp;amp;nbsp; &amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;nbsp; - Deploying Co-Process (1/1) (attempt #0) to
&gt; &amp;gt; e63d829deafc144cd82efd73979dd056 @ 083f69d029de (dataPort=35758)
&gt; &amp;gt; 2020-07-17 09:51:48,955 INFO&amp;amp;nbsp;
&gt; &amp;gt; 
org.apache.flink.runtime.executiongraph.ExecutionGraph&amp;amp;nbsp;
&gt; &amp;amp;nbsp; &amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;nbsp; - Co-Process -&amp;amp;gt; (Sink: Unnamed, Sink: 
Unnamed) (1/1)
&gt; &amp;gt; (618b75fcf5ea05fb5c6487bec6426e31) switched from SCHEDULED to
&gt; DEPLOYING.
&gt; &amp;gt; 2020-07-17 09:51:48,955 INFO&amp;amp;nbsp;
&gt; &amp;gt; 
org.apache.flink.runtime.executiongraph.ExecutionGraph&amp;amp;nbsp;
&gt; &amp;amp;nbsp; &amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;nbsp; - Deploying Co-Process -&amp;amp;gt; (Sink: 
Unnamed, Sink:
&gt; Unnamed) (1/1)
&gt; &amp;gt; (attempt #0) to e63d829deafc144cd82efd73979dd056 @ 083f69d029de
&gt; &amp;gt; (dataPort=35758)
&gt; &amp;gt; 2020-07-17 09:51:49,346 INFO&amp;amp;nbsp;
&gt; &amp;gt; 
org.apache.flink.runtime.executiongraph.ExecutionGraph&amp;amp;nbsp;
&gt; &amp;amp;nbsp; &amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;nbsp; - Co-Process -&amp;amp;gt; (Sink: Unnamed, Sink: 
Unnamed) (1/1)
&gt; &amp;gt; (618b75fcf5ea05fb5c6487bec6426e31) switched from DEPLOYING to 
RUNNING.
&gt; &amp;gt; 2020-07-17 09:51:49,370 INFO&amp;amp;nbsp;
&gt; &amp;gt; 
org.apache.flink.runtime.executiongraph.ExecutionGraph&amp;amp;nbsp;
&gt; &amp;amp;nbsp; &amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;nbsp; - Source: Custom Source (1/1)
&gt; (274b3df03e1fab627059c1a78e4a26da)
&gt; &amp;gt; switched from DEPLOYING to RUNNING.
&gt; &amp;gt; 2020-07-17 09:51:49,370 INFO&amp;amp;nbsp;
&gt; &amp;gt; 
org.apache.flink.runtime.executiongraph.ExecutionGraph&amp;amp;nbsp;
&gt; &amp;amp;nbsp; &amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;nbsp; - Source: Custom Source (1/1)
&gt; (141f0dc22b624b39e21127f637ba63c2)
&gt; &amp;gt; switched from DEPLOYING to RUNNING.
&gt; &amp;gt; 2020-07-17 09:51:49,377 INFO&amp;amp;nbsp;
&gt; &amp;gt; 
org.apache.flink.runtime.executiongraph.ExecutionGraph&amp;amp;nbsp;
&gt; &amp;amp;nbsp; &amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;nbsp; - Co-Process (1/1) 
(d0309f26a545e74643382ed3f758269b)
&gt; switched from
&gt; &amp;gt; DEPLOYING to RUNNING.
&gt; &amp;gt; 2020-07-17 09:51:49,377 INFO&amp;amp;nbsp;
&gt; &amp;gt; 
org.apache.flink.runtime.executiongraph.ExecutionGraph&amp;amp;nbsp;
&gt; &amp;amp;nbsp; &amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;nbsp; - Source: Custom Source -&amp;amp;gt; Filter (1/1)
&gt; &amp;gt; (1177659bff014e8dbc3f0508055d4307) switched from DEPLOYING to 
RUNNING.
&gt; &amp;gt; 2020-07-17 09:51:49,493 INFO&amp;amp;nbsp;
&gt; &amp;gt; 
org.apache.flink.runtime.executiongraph.ExecutionGraph&amp;amp;nbsp;
&gt; &amp;amp;nbsp; &amp;amp;nbsp;
&gt; &amp;gt; &amp;amp;nbsp; - Co-Process (1/1) 
(d0309f26a545e74643382ed3f758269b)
&gt; switched from
&gt; &amp;gt; RUNNING to FAILED.
&gt; &amp;gt; java.lang.Exception: Exception while creating
&gt; StreamOperatorStateContext.
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:191)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:255)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTask.initializeStateAndOpen(StreamTask.java:1006)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$0(StreamTask.java:454)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:94)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:449)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:461)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:707)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; org.apache.flink.runtime.taskmanager.Task.run(Task.java:532)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; java.lang.Thread.run(Thread.java:748)
&gt; &amp;gt; Caused by: org.apache.flink.util.FlinkException: Could not restore
&gt; keyed
&gt; &amp;gt; state backend for
&gt; &amp;gt; 
LegacyKeyedCoProcessOperator_65e7116c7aa972ad18a796ae22bd6327_(1/1)
&gt; from
&gt; &amp;gt; any of the 1 provided restore options.
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:304)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:131)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 ... 9 more
&gt; &amp;gt; Caused by: 
org.apache.flink.runtime.state.BackendBuildingException:
&gt; Caught
&gt; &amp;gt; unexpected exception.
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:336)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:548)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:288)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:142)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:121)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 ... 11 more
&gt; &amp;gt; Caused by: java.io.EOFException
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; java.io.DataInputStream.readFully(DataInputStream.java:197)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; java.io.DataInputStream.readFully(DataInputStream.java:169)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer.deserialize(BytePrimitiveArraySerializer.java:85)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restoreKVStateData(RocksDBFullRestoreOperation.java:221)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restoreKeyGroupsInStateHandle(RocksDBFullRestoreOperation.java:168)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restore(RocksDBFullRestoreOperation.java:151)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 at
&gt; &amp;gt;
&gt; 
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:279)
&gt; 
&amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
 ... 15 more

?????? Flink Cli ????????

回复