Hello :) We have a flink job (v 1.17) on k8s (using official flink-k8s-operator) that reads data from kafka and writes it to s3 using flink-sql using compaction. Job sometimes fails and continues from checkpoint just fine, but once a couple of days we experience a crash loop. Job cannot continue from the latest checkpoint and fails with such exception:

java.util.NoSuchElementException
	at java.base/java.util.ArrayList$Itr.next(Unknown Source)
	at org.apache.flink.connector.file.table.stream.compact.CompactOperator.initializeState(CompactOperator.java:114)
	at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.initializeOperatorState(StreamOperatorStateHandler.java:122)
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:274)
	at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:106)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:734)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:709)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:675)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:952)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:921)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:745)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562)
	at java.base/java.lang.Thread.run(Unknown Source)

Anyone experienced something like this? here’s the relevant code: https://github.com/apache/flink/blob/release-1.17/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/compact/CompactOperator.java#L114 It looks like

CompactOperator

is calling

next()

on iterator without checking

hasNext()

first - anyone know why? Why

context.getOperatorStateStore().getListState(metaDescriptor)

returns empty iterator? Is latest checkpoint broken in such case? We have identical job, but without compaction, and it works smoothly for a couple of weeks now. Here’s how the table look like. The whole job is just

select

from kafka and

insert

to s3.

CREATE EXTERNAL TABLE IF NOT EXISTS hive.`foo`.`bar` (
  `foo_bar1` STRING,
  `foo_bar2` STRING,
  `foo_bar3` STRING,
  `foo_bar4` STRING
  )
  PARTITIONED BY (`foo_bar1` STRING, `foo_bar2` STRING, `foo_bar3` STRING)
  STORED AS parquet
  LOCATION '<s3a://my/bucket/>'
  TBLPROPERTIES (
    'auto-compaction' = 'true',
    'compaction.file-size' = '128MB',
    'sink.parallelism' = '8',
    'format' = 'parquet',
    'parquet.compression' = 'SNAPPY',
    'sink.rolling-policy.rollover-interval' = '1 h',
    'sink.partition-commit.policy.kind' = 'metastore'
  )

Any help appreciated, thanks :)