hi hi all. Need help pls. Downgraded to 1.20.1, Iceberg 1.9.1 hoping my problems was version incompatibility, as all searches pointed at that... below is my dockerfile used to build the flink container being used. I've now added the jackson jar, as previous errors pointed/complaint about that being missing.

FROM arm64v8/flink:1.20.1-scala_2.12-java17
SHELL ["/bin/bash", "-c"]
ENV FLINK_HOME=/opt/flink
ENV HIVE_HOME=$FLINK_HOME/conf/
WORKDIR $FLINK_HOME


RUN echo "--> Setup Directory Structure" && \
    mkdir -p /opt/flink/conf/ && \
    mkdir -p /opt/flink/checkpoints && \
    mkdir -p /opt/flink/rocksdb 

RUN echo "--> Install JARs: Flink's S3 plugin" && \
    mkdir -p ./plugins/s3-fs-hadoop && \
    mv ./opt/flink-s3-fs-hadoop-1.20.1.jar ./plugins/s3-fs-hadoop/

RUN echo "--> Install JARs: => Generic"
COPY stage/bundle-2.31.9.jar                            /opt/flink/lib/bundle-2.31.9.jar  
COPY stage/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar    /opt/flink/lib/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar
COPY stage/jackson-core-2.20.1.jar                      /opt/flink/lib/jackson-core-2.20.1.jar
COPY stage/postgresql-42.7.6.jar                        /opt/flink/lib/postgresql-42.7.6.jar
# CDC Support <https://nightlies.apache.org/flink/flink-cdc-docs-release-3.4/docs/connectors/flink-sources/overview/#supported-flink-versions>
COPY stage/flink-sql-connector-postgres-cdc-3.5.0.jar   /opt/flink/lib/flink-sql-connector-postgres-cdc-3.5.0.jar

# Flink version specific
RUN echo "--> Install JARs: => Flink 1.20.1"
COPY stage/flink-sql-json-1.20.1.jar                    /opt/flink/lib/flink-sql-json-1.20.1.jar     
COPY stage/flink-sql-parquet-1.20.1.jar                 /opt/flink/lib/flink-sql-parquet-1.20.1.jar
COPY stage/flink-json-1.20.1.jar                        /opt/flink/lib/flink-json-1.20.1.jar
COPY stage/flink-connector-jdbc-3.3.0-1.20.jar          /opt/flink/lib/flink-connector-jdbc-3.3.0-1.20.jar

RUN echo "--> Install JARs: => Iceberg 1.9.1"
COPY stage/iceberg-flink-1.20-1.9.1.jar                 /opt/flink/lib/iceberg-flink-1.20-1.9.1.jar
COPY stage/iceberg-flink-runtime-1.20-1.9.1.jar         /opt/flink/lib/iceberg-flink-runtime-1.20-1.9.1.jar
COPY stage/iceberg-aws-bundle-1.9.1.jar                 /opt/flink/iceberg-aws-bundle-1.9.1.jar

RUN echo "--> Set Ownerships of /opt/flink" && \
    chown -R flink:flink $FLINK_HOME 

USER flink:flink

CMD ./bin/start-cluster.sh && sleep infinity

below is the polaris bits from my docker-compose file, included as this creates the polaris catalog, uses the previously created minio warehouse/iceberg bucket

polaris:
    image: apache/polaris:latest
    hostname: polaris
    container_name: polaris
    depends_on:
      postgrescat:
        condition: service_healthy
      minio:
        condition: service_healthy
      polaris-bootstrap:
        condition: service_completed_successfully

    ports:
      # API port
      - "8181:8181"
      # Management port (metrics and health checks)
      - "8182:8182"
      # Optional, allows attaching a debugger to the Polaris JVM
      - "5005:5005"
    environment:

      POLARIS_PERSISTENCE_TYPE: relational-jdbc
      POLARIS_PERSISTENCE_RELATIONAL_JDBC_MAX_RETRIES: 5
      POLARIS_PERSISTENCE_RELATIONAL_JDBC_INITIAL_DELAY_IN_MS: 100
      POLARIS_PERSISTENCE_RELATIONAL_JDBC_MAX_DURATION_IN_MS: 5000
      QUARKUS_DATASOURCE_JDBC_URL: jdbc:postgresql://${POSTGRES_CAT_HOST}:${POSTGRES_CAT_PORT}/${POLARIS_REALM}
      QUARKUS_DATASOURCE_USERNAME: ${POSTGRES_CAT_USER}
      QUARKUS_DATASOURCE_PASSWORD: ${POSTGRES_CAT_PASSWORD}
      POLARIS_REALM_CONTEXT_REALMS: ${POLARIS_REALM}
      QUARKUS_OTEL_SDK_DISABLED: true
      polaris.features."ALLOW_INSECURE_STORAGE_TYPES": "true"
      polaris.features."SUPPORTED_CATALOG_STORAGE_TYPES": "[\"FILE\",\"S3\"]"
      polaris.readiness.ignore-severe-issues: "true"
      
      # S3 endpoint for MinIO
      AWS_REGION: ${AWS_REGION}
      AWS_ACCESS_KEY_ID: ${MINIO_ROOT_USER}
      AWS_SECRET_ACCESS_KEY: ${MINIO_ROOT_PASSWORD}
      POLARIS_BOOTSTRAP_CREDENTIALS: ${POLARIS_REALM},${ROOT_CLIENT_ID},${ROOT_CLIENT_SECRET}
      polaris.realm-context.realms: ${POLARIS_REALM}
      quarkus.otel.sdk.disabled: "true"

    volumes:
      - ./data/polaris:/var/lib/polaris
      - ./conf/polaris:/polaris 
    healthcheck:
      test: ["CMD", "curl", "<http://localhost:8182/q/health>"]
      interval: 5s
      timeout: 10s
      retries: 5
      start_period: 30s
    restart: unless-stopped

  polaris-bootstrap:
    image: apache/polaris-admin-tool:latest
    hostname: polaris-bootstrap
    container_name: polaris-bootstrap
    depends_on:
      postgrescat:
        condition: service_healthy    
    environment:
      - POLARIS_PERSISTENCE_TYPE=relational-jdbc
      - QUARKUS_DATASOURCE_JDBC_URL=jdbc:postgresql://${POSTGRES_CAT_HOST}:${POSTGRES_CAT_PORT}/${POLARIS_REALM}
      - QUARKUS_DATASOURCE_USERNAME=${POSTGRES_CAT_USER}
      - QUARKUS_DATASOURCE_PASSWORD=${POSTGRES_CAT_PASSWORD}
      - POLARIS_REALM=${POLARIS_REALM}
      - ROOT_CLIENT_ID=${ROOT_CLIENT_ID}
      - ROOT_CLIENT_SECRET=${ROOT_CLIENT_SECRET}
    command:
      - "bootstrap"
      - "--realm=${POLARIS_REALM}"
      - "--credential=${POLARIS_REALM},${ROOT_CLIENT_ID},${ROOT_CLIENT_SECRET}"

  polaris-setup:
    image: alpine/curl
    hostname: polaris-setup
    container_name: polaris-setup
    depends_on:
      polaris:
        condition: service_healthy
    environment:
      - CLIENT_ID=${ROOT_CLIENT_ID}
      - CLIENT_SECRET=${ROOT_CLIENT_SECRET}
      - CATALOG_NAME=${CATALOG_NAME}
      - POLARIS_REALM=${POLARIS_REALM}
      - MINIO_BUCKET=${MINIO_BUCKET}
      - MINIO_ENDPOINT=${MINIO_ENDPOINT}
    volumes:
      - ./conf/polaris/:/polaris
    entrypoint: "/bin/sh"
    command:
      - "-c"
      - >-
        chmod +x /polaris/create-catalog.sh;
        chmod +x /polaris/obtain-token.sh;
        source /polaris/obtain-token.sh $$POLARIS_REALM;
        echo Creating catalog: $$CATALOG_NAME in realm: $$POLARIS_REALM;
        export STORAGE_CONFIG_INFO='{"storageType":"S3",
          "endpoint":"<http://localhost:9000>",
          "endpointInternal":"'$$MINIO_ENDPOINT'",
          "pathStyleAccess":true}';
        export STORAGE_LOCATION="s3://$$MINIO_BUCKET/iceberg";
        /polaris/create-catalog.sh $$POLARIS_REALM $$CATALOG_NAME $$TOKEN;
        echo Extra grants...;
        curl -H "Authorization: Bearer $$TOKEN" -H 'Content-Type: application/json' \
          -X PUT \
          <http://polaris:8181/api/management/v1/catalogs/${CATALOG_NAME}/catalog-roles/catalog_admin/grants> \
          -d '{"type":"catalog", "privilege":"CATALOG_MANAGE_CONTENT"}';
        echo Polaris Setup Complete.;

c_iceberg catalog create

CREATE CATALOG c_iceberg WITH (
   'type'='iceberg'
  ,'catalog-type'='rest'
  ,'uri'='<http://polaris:8181/api/catalog>'
  ,'warehouse'='icebergcat'
  ,'oauth2-server-uri'='<http://polaris:8181/api/catalog/v1/oauth/tokens>'
  ,'credential'='root:s3cr3t'
  ,'scope'='PRINCIPAL_ROLE:ALL'
  ,'s3.endpoint'='<http://minio:900>'
  ,'s3.access-key-id'='mnadmin'
  ,'s3.secret-access-key'='mnpassword'
  ,'s3.path-style-access'='true'
);


USE CATALOG c_iceberg;
-- Create using below
CREATE DATABASE IF NOT EXISTS c_iceberg.finflow;

flink sql being executed

CREATE OR REPLACE TABLE c_cdcsource.demog.accountholders (
     _id                BIGINT                  NOT NULL
    ,nationalid         VARCHAR(16)             NOT NULL
    ,firstname          VARCHAR(100)
    ,lastname           VARCHAR(100)
    ,dob                VARCHAR(10) 
    ,gender             VARCHAR(10)
    ,children           INT
    ,emailaddress       VARCHAR(100)
    ,mobilephonenumber  VARCHAR(20)
    ,accounts           STRING
    ,created_at         TIMESTAMP_LTZ(3)
    ,WATERMARK          FOR created_at AS created_at - INTERVAL '15' SECOND
    ,PRIMARY KEY (_id) NOT ENFORCED
) WITH (
     'connector'                           = 'postgres-cdc'
    ,'hostname'                            = 'postgrescdc'
    ,'port'                                = '5432'
    ,'username'                            = 'dbadmin'
    ,'password'                            = 'dbpassword'
    ,'database-name'                       = 'demog'
    ,'schema-name'                         = 'public'
    ,'table-name'                          = 'accountholders'
    ,'slot.name'                           = 'accountholders0'
    -- experimental feature: incremental snapshot (default off)
    ,'scan.incremental.snapshot.enabled'   = 'true'               -- experimental feature: incremental snapshot (default off)
    ,'scan.startup.mode'                   = 'initial'            -- <https://nightlies.apache.org/flink/flink-cdc-docs-release-3.1/docs/connectors/flink-sources/postgres-cdc/#startup-reading-position>     ,'decoding.plugin.name'                = 'pgoutput'
    ,'decoding.plugin.name'                = 'pgoutput'
);

SET 'execution.checkpointing.interval'   = '60s';
SET 'table.exec.sink.upsert-materialize' = 'NONE';
SET 'pipeline.name' = 'Persist into Iceberg: accountholders table';
CREATE OR REPLACE TABLE c_iceberg.finflow.accountholders WITH (
     'file.format'                       = 'parquet'
    ,'compaction.min.file-num'           = '2'
    ,'compaction.early-max.file-num'     = '50'
    ,'snapshot.time-retained'            = '1h'
    ,'snapshot.num-retained.min'         = '5'
    ,'snapshot.num-retained.max'         = '20'
    ,'table.exec.sink.upsert-materialize'= 'NONE'
) AS 
SELECT * FROM c_cdcsource.demog.accountholders;

The above resulted in the below

[ERROR] Could not execute SQL statement. Reason:
org.apache.iceberg.exceptions.NoSuchNamespaceException: Namespace does not exist: finflow

After I got the above namespace problem I executed the following

curl -X POST <http://localhost:8181/api/catalog/v1/icebergcat/namespaces> \
    -H "Authorization: Bearer ${TOKEN}" \
    -H 'Content-Type: application/json' \
    -d '{"namespace": ["finflow"], "properties": {"description": "Iceberg catalog database"}}' | jq

I then reran my above CTAS errors below that

[ERROR] Could not execute SQL statement. Reason:
java.lang.AbstractMethodError: Receiver class org.apache.iceberg.flink.FlinkCreateTableOptions$$Lambda$1155/0x0000000601838000 does not define or inherit an implementation of the resolved method 'abstract void generate(org.apache.iceberg.shaded.com.fasterxml.jackson.core.JsonGenerator)' of interface org.apache.iceberg.util.JsonUtil$ToJson.

Flink SQL>