hi hi all... bit stuck, trying to push data into a iceberg tables. Need someone to see where I went wrong... missed something the below is the code that created the bucket, I can see the bucket in MinIO: CDC Source table

# CDC Sources
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'
);

flink 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;

contents from .env file driving the compose.yaml ROOT_CLIENT_ID=root ROOT_CLIENT_SECRET=s3cr3t CATALOG_NAME=icebergcat POLARIS_REALM=findept MINIO_ROOT_USER=mnadmin MINIO_ROOT_PASSWORD=mnpassword MINIO_ALIAS=minio MINIO_ENDPOINT=http://minio:9000 MINIO_BUCKET=warehouse AWS_ACCESS_KEY_ID=mnadmin AWS_SECRET_ACCESS_KEY=mnpassword AWS_REGION=za-south-1 AWS_DEFAULT_REGION=za-south-1 polaris-bootsrap service from docker-compose

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 service from docker-compose.

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.;

output from fetch catalogs api call, verifying the catalog and full path

curl -X GET <http://localhost:8181/api/management/v1/catalogs> \
>   -H "Authorization: Bearer ${TOKEN}" | jq
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   388  100   388    0     0  13756      0 --:--:-- --:--:-- --:--:-- 13857
{
  "catalogs": [
    {
      "type": "INTERNAL",
      "name": "icebergcat",
      "properties": {
        "default-base-location": "<s3://warehouse/iceberg>"
      },
      "createTimestamp": 1765474601004,
      "lastUpdateTimestamp": 1765474601004,
      "entityVersion": 1,
      "storageConfigInfo": {
        "endpoint": "<http://localhost:9000>",
        "endpointInternal": "<http://minio:9000>",
        "pathStyleAccess": true,
        "storageType": "S3",
        "allowedLocations": [
          "<s3://warehouse/iceberg>"
        ]
      }
    }
  ]
}

ERROR

> 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;
[ERROR] Could not execute SQL statement. Reason:
java.lang.AbstractMethodError: Receiver class org.apache.iceberg.flink.FlinkCreateTableOptions$$Lambda$1164/0x0000000801826a00 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.

docker images

arm64v8/flink:1.20.2-scala_2.12-java17                      79a54781f5d0        789MB             0B        
georgelza/apacheflink-base-1.20.2-scala_2.12-java17:1.0.2   dd8585afe66c       6.84GB             0B    U

Flink container build / Dockerfile, as ... error search is saying version incompatibility...

FROM arm64v8/flink:1.20.2-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.2.jar ./plugins/s3-fs-hadoop/

RUN echo "--> Install Flink 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/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
COPY stage/flink-sql-json-1.20.2.jar                    /opt/flink/lib/flink-sql-json-1.20.2.jar     
COPY stage/flink-sql-parquet-1.20.2.jar                 /opt/flink/lib/flink-sql-parquet-1.20.2.jar
COPY stage/flink-json-1.20.2.jar                        /opt/flink/lib/flink-json-1.20.2.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 Flink JARs: => Open Table Formats"
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