使用Flink Kubernetes Operator部署带MinIO S3的SessionJob遇错求助

在本地k3d环境中，基于Flink Kubernetes Operator（release-1.4.0版本）搭建Session集群，计划用MinIO提供的S3存储实现Checkpoint和Savepoint，MinIO Pod已部署完成。

自定义镜像构建操作 #

基于官方Operator镜像构建自定义镜像，执行以下操作：

• 在/opt/flink/plugins目录下创建s3-fs-hadoop和s3-fs-presto子目录
• 将镜像内/opt/flink/opt目录中的对应版本JAR文件复制到上述子目录中

集群启动日志 #

创建FlinkDeployment后，Pod日志显示S3插件已启用，但同时出现只读文件系统相关警告：

Enabling required built-in plugins
Linking flink-s3-fs-hadoop-1.16.1.jar to plugin directory
Linking1 flink-s3-fs-hadoop-1.16.1 to plugin directory
Successfully enabled flink-s3-fs-hadoop-1.16.1.jar
Linking flink-s3-fs-presto-1.16.1.jar to plugin directory
Linking1 flink-s3-fs-presto-1.16.1 to plugin directory
Successfully enabled flink-s3-fs-presto-1.16.1.jar
sed: couldn't open temporary file /opt/flink/conf/sedNLg8lR: Read-only file system
sed: couldn't open temporary file /opt/flink/conf/sed6vCEtB: Read-only file system
/docker-entrypoint.sh: line 74: /opt/flink/conf/flink-conf.yaml: Read-only file system
/docker-entrypoint.sh: line 90: /opt/flink/conf/flink-conf.yaml.tmp: Read-only file system
Starting kubernetes-session as a console application on host my-flink-cluster-5967dc856b-rlznq

FlinkDeployment配置清单 #

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: my-flink-cluster
spec:
  image: k3d-test-app-registry:5050/flink-kube-s3:v1 
  flinkVersion: v1_16
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
    kubernetes.operator.periodic.savepoint.interval: 1h
    state.savepoints.dir: "s3://state/minio-service:9000"
    state.checkpoints.dir: "s3://state/minio-service:9000"
    s3.endpoint: http://minio-service:9000
    s3.path.style.access: "true"
    s3.access-key: ****
    s3.secret-key: ****
    metrics.reporters: prom
    metrics.reporter.prom.factory.class: org.apache.flink.metrics.prometheus.PrometheusReporterFactory
  serviceAccount: flink
  jobManager:
    resource:
      memory: "1024m"
      cpu: 1
  taskManager:
    resource:
      memory: "1024m"
      cpu: 1
  podTemplate:
    spec:
      containers:
        - name: flink-main-container 
          env:
          - name: FLINK_PARALLELISM  
            value: "3"  
          - name: CHECKPOINT_DIR  
            value: "s3://state/minio-service:9000"  
          - name: ENABLE_BUILT_IN_PLUGINS
            value: flink-s3-fs-hadoop-1.16.1.jar;flink-s3-fs-presto-1.16.1.jar    

FlinkSessionJob配置清单 #

部署官方示例作业时使用以下配置：

apiVersion: flink.apache.org/v1beta1
kind: FlinkSessionJob
metadata:
  name: my-flink-job
spec:
  deploymentName: my-flink-cluster
  flinkConfiguration:
    kubernetes.operator.periodic.savepoint.interval: 1h
    state.savepoints.dir: "s3://state/minio-service:9000"
    state.checkpoints.dir: "s3://state/minio-service:9000"
    s3.endpoint: http://minio-service:9000
    s3.path.style.access: "true"
    s3.access-key: ****
    s3.secret-key: ****
  job:
    jarURI: https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming_2.12/1.16.1/flink-examples-streaming_2.12-1.16.1-TopSpeedWindowing.jar
    parallelism: 2
    upgradeMode: stateless

作业启动错误日志 #

部署SessionJob后，FlinkDeployment Pod日志出现以下错误：

Caused by: org.apache.flink.util.FlinkRuntimeException: Failed to create checkpoint storage at checkpoint coordinator side.
        at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.<init>(CheckpointCoordinator.java:337) ~[flink-dist-1.16.1.jar:1.16.1]
        at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.<init>(CheckpointCoordinator.java:245) ~[flink-dist-1.16.1.jar:1.16.1]
        at org.apache.flink.runtime.executiongraph.DefaultExecutionGraph.enableCheckpointing(DefaultExecutionGraph.java:511) ~[flink-dist-1.16.1.jar:1.16.1]
        at org.apache.flink.runtime.executiongraph.DefaultExecutionGraphBuilder.buildGraph(DefaultExecutionGraphBuilder.java:317) ~[flink-dist-1.16.1.jar:1.16.1]
        at org.apache.flink.runtime.scheduler.DefaultExecutionGraphFactory.createAndRestoreExecutionGraph(DefaultExecutionGraphFactory.java:156) ~[flink-dist-1.16.1.jar:1.16.1]
        at org.apache.flink.runtime.scheduler.SchedulerBase.createAndRestoreExecutionGraph(SchedulerBase.java:361) ~[flink-dist-1.16.1.jar:1.16.1]
        at org.apache.flink.runtime.scheduler.SchedulerBase.<init>(SchedulerBase.java:206) ~[flink-dist-1.16.1.jar:1.16.1]
        at org.apache.flink.runtime.scheduler.DefaultScheduler.<init>(DefaultScheduler.java:134) ~[flink-dist-1.16.1.jar:1.16.1]
        at org.apache.flink.runtime.scheduler.DefaultSchedulerFactory.createInstance(DefaultSchedulerFactory.java:152) ~[flink-dist-1.16.1.jar:1.16.1]
        at org.apache.flink.runtime.jobmaster.DefaultSlotPoolServiceSchedulerFactory.createScheduler(DefaultSlotPoolServiceSchedulerFactory.java:119) ~[flink-dist-1.16.1.jar:1.16.1]
        at org.apache.flink.runtime.jobmaster.JobMaster.createScheduler(JobMaster.java:369) ~[flink-dist-1.16.1.jar:1.16.1]
        at org.apache.flink.runtime.jobmaster.JobMaster.<init>(JobMaster.java:346) ~[flink-dist-1.16.1.jar:1.16.1]
        at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.internalCreateJobMasterService(DefaultJobMasterServiceFactory.java:123) ~[flink-dist-1.16.1.jar:1.16.1]
        at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.lambda$createJobMasterService$0(DefaultJobMasterServiceFactory.java:95) ~[flink-dist-1.16.1.jar:1.16.1]
        at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedSupplier$4(FunctionUtils.java:112) ~[flink-dist-1.16.1.jar:1.16.1]
        at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1604) ~[?:1.8.0_372]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_372]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_372]
        ... 1 more
Caused by: org.apache.flink.core.fs.UnsupportedFileSystemSchemeException: 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/.

请求 #

错误提示S3插件缺失，但实际Pod内plugins目录已存在对应JAR文件，尝试过相关解决方案仍无效，请求排查操作问题或提供FlinkSessionJob结合S3存储的可用示例。

内容的提问来源于stack exchange，提问作者ypanag