with table_result.collect() as results:
    for result in results:
        print(result)

Message: Forbidden!Configured service account doesn't have access. Service account may have been revoked. deployments.extensions "eventlogger-pipeline-v0" is forbidden: User "system:serviceaccount:eventlogger-pipeline-staging:flink" cannot get resource "deployments" in API group "extensions" in the namespace "eventlogger-pipeline-staging"

eventlogger-pipeline-v0

eventlogger-pipeline-v0-1

env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

2023-01-06T21:10:11.885Z {"applicationARN":"arn:aws:kinesisanalytics:us-east-1:542230711021:application/laap-kda-ue1-con-sr-sr-per-dlvry-metrc-imprsn-prod","applicationVersionId":"15","locationInformation":"org.apache.flink.runtime.io.network.netty.PartitionRequestQueue.handleException(PartitionRequestQueue.java:287)","logger":"org.apache.flink.runtime.io.network.netty.PartitionRequestQueue","message":"Encountered error while consuming partitions","messageSchemaVersion":"1","messageType":"ERROR","threadName":"Flink Netty Server (6121) Thread 0","throwableInformation":"org.apache.flink.shaded.netty4.io.netty.channel.unix.Errors$NativeIoException: readAddress(..) failed: Connection reset by peer\n"}

high-availability.storageDir

flink-conf.yaml.

Could not find a file system implementation for scheme 's3'. The scheme is directly supported by Flink through the following plugin(s): flink-s3-fs-hadoop, flink-s3-fs-presto. Please ensure that each plugin resides within its own subfolder within the plugins directory. See <https://nightlies.apache.org/flink/flink-docs-stable/docs/deployment/filesystems/plugins/> for more information. If you want to use a Hadoop file system for that scheme, please add the scheme to the configuration fs.allowed-fallback-filesystems. For a full list of supported file systems, please see <https://nightlies.apache.org/flink/flink-docs-stable/ops/filesystems/>.

yield output_tag, value

WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.apache.flink.api.java.ClosureCleaner (file:/opt/flink/lib/flink-dist-1.16.1.jar) to field java.lang.String.value
WARNING: Please consider reporting this to the maintainers of org.apache.flink.api.java.ClosureCleaner
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release

-- when i execute this sql, i can get the correct result
SELECT IF(`marketing_flow_id` IS NOT NULL, `marketing_flow_id`, '') FROM source;

-- when i execute this sql, i get the error result(return  blank)
SELECT IF(`marketing_flow_id` IS  NULL, '', `marketing_flow_id`) FROM source;

Tuple2<String, String>

public static class MyKeyedProcessFunction
      extends KeyedProcessFunction<String, Tuple2<String, Integer>, Tuple2<String, Integer>> {
    private ValueState<Integer> state;

    @Override
    public void open(Configuration parameters) throws Exception {
      state = getRuntimeContext().getState(new ValueStateDescriptor<>("sum", Integer.class));
    }

    @Override
    public void processElement(
        Tuple2<String, Integer> value,
        Context ctx,
        Collector<Tuple2<String, Integer>> out) throws Exception {
      Integer sum = state.value();
      if (sum == null) {
        sum = 0;
      }
      sum += value.f1;
      state.update(sum);
      ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() + 5000);
    }

    @Override
    public void onTimer(
        long timestamp,
        OnTimerContext ctx,
        Collector<Tuple2<String, Integer>> out) throws Exception {
      out.collect(Tuple2.of(ctx.getCurrentKey(), state.value()));
      state.clear();
    }
  }

onTimer

aaa,50
aaa,40
aaa,10

(aaa,100)
(aaa, null)
(aaa, null)

(aaa,100)

E1("A",1,...) -> E2("B",1,...) -> E3("C",1,...)

sequence_number

AUTO_INCREMENT

state.checkpoints.num-retained: "5"