Flink问题之无法启动如何解决

问题一：Flink savepoint迁移问题

社区各位大佬，我遇到了一个Flink版本升级后，savepoint不会正确恢复问题。

版本从1.9.3 升级到1.11.0或1.11.3.

连接器使用的pulsar的连接器。在Source的checkpoint数据结构如下

oldUnionOffsetStates = stateStore.getUnionListState(

new ListStateDescriptor<>(

OFFSETS_STATE_NAME,

TypeInformation.of(new TypeHint>() {

})));

oldUnionSubscriptionNameStates =

stateStore.getUnionListState(

new ListStateDescriptor<>(

OFFSETS_STATE_NAME + "_subName",

TypeInformation.of(new TypeHint () {

})));

我在本地通过bin/flink savepoint 命令保存下1.9.3的state，然后停止flink 1.9.3的集群，启动1.11.3的flink集群，使用bin/flink run -s 参数恢复任务。

任务在启动后，会遇到下面的错误

2021-03-11 10:02:25

java.lang.Exception: Exception while creating StreamOperatorStateContext.

at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:222)

at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:248)

at org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:290)

at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$1(StreamTask.java:506)

at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:92)

at org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:475)

at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:526)

at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721)

at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546)

at java.base/java.lang.Thread.run(Thread.java:832)

Caused by: org.apache.flink.util.FlinkException: Could not restore operator state backend for StreamSource_15541fe0b20ca4f01df784b178a8cc9d_(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.operatorStateBackend(StreamTaskStateInitializerImpl.java:283)

at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:156)

... 9 more

Caused by: org.apache.flink.runtime.state.BackendBuildingException: Failed when trying to restore operator state backend

at org.apache.flink.runtime.state.DefaultOperatorStateBackendBuilder.build(DefaultOperatorStateBackendBuilder.java:86)

at org.apache.flink.runtime.state.memory.MemoryStateBackend.createOperatorStateBackend(MemoryStateBackend.java:314)

at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$operatorStateBackend$0(StreamTaskStateInitializerImpl.java:274)

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.iohttp://java.io.EOFException: No more bytes left.

at org.apache.flink.api.java.typeutils.runtime.NoFetchingInput.require(NoFetchingInput.java:79)

at com.esotericsoftware.kryo.iohttp://com.esotericsoftware.kryo.io.Input.readVarInt(Input.java:355)

at com.esotericsoftware.kryo.iohttp://com.esotericsoftware.kryo.io.Input.readInt(Input.java:350)

at com.esotericsoftware.kryo.serializers.UnsafeCacheFields$UnsafeIntField.read(UnsafeCacheFields.java:46)

at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)

at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)

at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:346)

at org.apache.flink.api.java.typeutils.runtime.TupleSerializer.deserialize(TupleSerializer.java:151)

at org.apache.flink.api.java.typeutils.runtime.TupleSerializer.deserialize(TupleSerializer.java:37)

at org.apache.flink.runtime.state.OperatorStateRestoreOperation.deserializeOperatorStateValues(OperatorStateRestoreOperation.java:191)

at org.apache.flink.runtime.state.OperatorStateRestoreOperation.restore(OperatorStateRestoreOperation.java:165)

at org.apache.flink.runtime.state.DefaultOperatorStateBackendBuilder.build(DefaultOperatorStateBackendBuilder.java:83)

... 15 more

请问题大佬们可以提供排查问题的办法或者解决方案吗？*来自志愿者整理的flink邮件归档

参考回答：

新的集群使用的是更新之后的 pulsar connector ？我看了下 pulsar-flink 的代码，这个更新对 state 是破坏性的。

• unionOffsetStates = stateStore.getUnionListState(
• new ListStateDescriptor<>(
• OFFSETS_STATE_NAME,
• TypeInformation.of(new TypeHint>() {
• })));

解决方法 :? 1. 尝试通过 state-processor-api 重写下 state ? 2. 1.9 的 pulsar-flink connector 可以跑在 1.11 的 flink 上吗？

感觉后面还有不兼容的更新

new ListStateDescriptor<>( OFFSETS_STATE_NAME, - TypeInformation.of(new TypeHint>() { + TypeInformation.of(new TypeHint>() { })));

不兼容的 state 更新，如果不好做 schema migration，可以试试 StatefulSinkWriterOperator 的方法：同时读取新旧 state + 只写新 state。

可以等 streamnative 的人确认下。

关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/359397?spm=a2c6h.13262185.0.0.54e839c0LrOat6

问题二：将每个tm的slot数从2降低到1，任务反而无法启动怎么办

您好： 任务的拓扑图如下，parallelism的设置是140，但是中间有个操作的并行度设置成了50。 集群剩余的slot总数是195。 如果将每个tm的slot数设置为2，任务能够正常启动，并且分配了70个tm和140个slot，符合预期。 可如果将每个tm的slot数设置为1，便只分配了115个slot。任务会卡在creating状态，并且几分钟后，会抛出异常，“Could not allocate all requires slots within timeout of 300000 ms. Slots required: 470, slots allocated: 388” 这里面有几个问题： 1. 将slot数设置为1后，异常中提示“Slots required: 470”，这个470似乎完全没有考虑slot share（我们并没有手动设置SlotSharingGroup）。这是为啥？ 2. 将slot数设置为1后，异常中提示“slots allocated: 388”，而整个集群剩余的slot其实只有195个，这个388怎么来的？ 3. 最大的并行度应是140，为何只分配了115个slot呢？ 我们使用的flink版本是1.6.2。 期待你们的回复~*来自志愿者整理的flink邮件归档

参考回答：

建议看下集群剩余的内存情况，看是不是 140 个 TAskManager 内存不够了

关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/359418?spm=a2c6h.13262185.0.0.54e839c0YhZuwm

问题三：提交两个SQL任务，其中一个不生效怎么办？

大家好，我通过Yarn Per Job模式提交了两个任务，后一个提交的任务会变成前一个一样的任务，而且好像不产生实际作用(?)，看起来像是在给前一个做HA，不知道是不是什么配置没搞好呢？

两个任务分别是 Kafka Topic1 -> Kafka Topic2 和 Kafka Topic2 -> Postgre，两个任务中的Topic2用的是同一个。

之前也尝试过，Topic1同时写入Topic2和Postgre，好像也同样不会生效，这是为什么呢？

谢谢大家。*来自志愿者整理的flink邮件归档

参考回答：

多个insert的话要用statementset去提交

关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/359419?spm=a2c6h.13262185.0.0.54e839c0YhZuwm

问题四：MapState 无法更新怎么办

主要代码如下：

class getRule extends KeyedBroadcastProcessFunction[String, KafkaStreamSource, List[Rule], KafkaStreamSource] { private var carEfenceState: MapState[String, Boolean] = _

override def open(parameters: Configuration): Unit = {

carEfenceState = getRuntimeContext.getMapState(new

MapStateDescriptor[String, Boolean]("carEfenceState", classOf[String],

classOf[Boolean]))

}

override def processBroadcastElement(in2: List[Rule], context:

KeyedBroadcastProcessFunction[String, KafkaStreamSource, List[Rule],

KafkaStreamSource]#Context, collector: Collector[KafkaStreamSource]): Unit =

{

context.getBroadcastState(ruleStateDescriptor).put("rules", in2)

}

override def processElement(kafkaSource: KafkaStreamSource,

readOnlyContext: KeyedBroadcastProcessFunction[String, KafkaStreamSource,

List[Rule], KafkaStreamSource]#ReadOnlyContext, collector:

Collector[KafkaStreamSource]): Unit = {

val ruleIterator =

readOnlyContext.getBroadcastState(ruleStateDescriptor).immutableEntries().iterator()

while (ruleIterator.hasNext) {

val ruleMap: Map.Entry[String, List[Rule]] = ruleIterator.next()

val ruleList: List[Rule] = ruleMap.getValue

for (rule <- ruleList) {

val mapKey = kafkaSource.vno + rule.id

val tempState = carEfenceState.get(mapKey)

val currentState = if (tempState != null) tempState else false

// 业务逻辑

if (!currentState) {

...

carEfenceState.put(mapKey, true)

...

} else if (currentState) {

...

carEfenceState.remove(mapKey)

...

}

}

}

}

}

c*来自志愿者整理的flink邮件归档

参考回答：

A read-only view of the {@link BroadcastState}.

*

• Although read-only, the user code should not modify the value returned by the {@link
  
• #get(Object)} or the entries of the immutable iterator returned by the {@link
  
• #immutableEntries()}, as this can lead to inconsistent states. The reason for this is that we do
  
• not create extra copies of the elements for performance reasons.
  

*

• @param The key type of the elements in the {@link ReadOnlyBroadcastState}.
  
• @param The value type of the elements in the {@link ReadOnlyBroadcastState}.
  

*/

这是源码中对ReadOnlyBroadcastState的描述，希望对你有帮助

关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/359504?spm=a2c6h.13262185.0.0.133c39c0Clkic3

问题五：BUG ：DataStream 转 Table 后无法 触发窗口计算怎么办

1：当DataStream是由 一个table 经过 group by rowtime 转换过来的就无法触发窗口

例如： tableEnv.createTemporaryView("test3", tableEnv.sqlQuery("select msg,rowtime from test group by msg,rowtime"));

// 获得 DataStream，并定义wtm生成 SingleOutputStreamOperator r = tableEnv.toRetractStream(tableEnv.from("test3"), Row.class) .filter(x -> x.f0) // map ........ .returns(Types.TUPLE(Types.STRING, Types.LONG)) .assignTimestampsAndWatermarks( WatermarkStrategy.>forBoundedOutOfOrderness(Duration.ofSeconds(1)) .withTimestampAssigner(((element, recordTimestamp) -> element.f1)) );

参考 官方文档： https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/streaming/time_attributes.html

// stream - 转 Table，指定Rowtime tableEnv.createTemporaryView("test5", r, $("msg"), $("rowtime").rowtime());

String sql5 = "select " + "msg," + "count(1) cnt" + " from test5 " + " group by TUMBLE(rowtime, INTERVAL '30' SECOND), msg " + ""; tableEnv.executeSql("insert into printlnRetractSink " + sql5);

结果： 无法触发窗口操作。 查调试源码： org.apache.flink.table.runtime.operators.window.WindowOperator // 返回的wtm永远都是 -9223372036854775808 public long getCurrentWatermark() { return internalTimerService.currentWatermark(); }

// 查看任务，watermark是正常在生成的。InternalTimerServiceImpl.advanceWatermark是正常为currentWatermark赋值。但是 internalTimerService.currentWatermark() 却拿的是-9223372036854775808

// 当 tableEnv.createTemporaryView("test3", tableEnv.sqlQuery("select msg,rowtime from test group by msg,rowtime")); 语句改为 tableEnv.createTemporaryView("test3", tableEnv.sqlQuery("select msg,rowtime from test"));

结果就是正确的。 所以这是一个bug吗？？*来自志愿者整理的flink邮件归档

参考回答：

你好， 你的flink版本是多少？ 之前有个bug是Table转datastream 会丢rowtime问题，看起来是这个问题。

我在[1]里修复了，你可以升级对应的版本试下。

祝好， Leonard [1]https://issues.apache.org/jira/browse/FLINK-21013 https://issues.apache.org/jira/browse/FLINK-21013

关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/359506?spm=a2c6h.13262185.0.0.133c39c0Clkic3