Мое потоковое приложение flink (v1.14.4) содержит коннектор JDBC, используемый для начальной выборки данных с сервера MySQL. Логика:
Когда я запускаю приложение локально, я вижу следующее исключение
14:52:00.401 [Source: TableSourceScan(table=[[default_catalog, default_database, asset, project=[id, status]]], fields=[id, status]) -> Calc(select=[CAST(_UTF-16LE'sergey-test-asset-id':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS id, status], where=[(id = _UTF-16LE'sergey-test-asset-id')]) -> TableToDataSteam(type=ROW<`id` STRING, `status` STRING> NOT NULL, rowtime=false) -> Sink: Print to Std. Out (4/4)#0] INFO o.a.flink.runtime.taskmanager.Task - Source: TableSourceScan(table=[[default_catalog, default_database, asset, project=[id, status]]], fields=[id, status]) -> Calc(select=[CAST(_UTF-16LE'sergey-test-asset-id':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS id, status], where=[(id = _UTF-16LE'sergey-test-asset-id')]) -> TableToDataSteam(type=ROW<`id` STRING, `status` STRING> NOT NULL, rowtime=false) -> Sink: Print to Std. Out (4/4)#0 (e8870cf296ac770346384fe2529b325f) switched from RUNNING to FINISHED.
...
14:57:52.963 [Checkpoint Timer] INFO o.a.f.r.c.CheckpointCoordinator - Failed to trigger checkpoint for job 8303c423dd7b9e3f303f0b299d7d37bb because Some tasks of the job have already finished and checkpointing with finished tasks is not enabled. Failure reason: Not all required tasks are currently running.
Я понимаю, что после оператора SQL select flink пометит оператор jdbc в состояние FINISHED, но мне нужно продолжить выполнение потокового приложения и иметь контрольные точки во время выполнения.
Нужно ли мне использовать свойство execution.checkpointing.checkpoints-after-tasks-finish.enabled: true
, чтобы решить мою проблему, ИЛИ мне нужно изменить граф приложения?
Да, нужно использовать
execution.checkpointing.checkpoints-after-tasks-finish.enabled: true
так что задание может поставить контрольную точку, несмотря на то, что исходные задачи jdbc завершены.