hello team , Objective: we are using flink with hu...
# troubleshooting
c
hello team , Objective: we are using flink with hudi to generate parquet files to push to s3/azure container[blob or adls gen2] details: infra : k8s FLINK_VERSION=1.17.1 HADOOP_VERSION=3.3.5 jars added to flink lib : hadoop-common, hadoop-aws, aws-java-sdk-s3, s3-fs-hadoop, azure-fs-hadoop Currently using s3-fs-hadoop [s3a], we are able to commit checkpoint and savepoint metadata to s3. We handle this with a combination of service account + aws oidc + s3 policy However the same doesn't seem to work for commiting .hoodi files.
Copy code
... 42 more
Caused by: java.nio.file.AccessDeniedException: <s3a://clen/results-store/.hoodie>: getFileStatus on <s3a://clen/results-store/.hoodie>: com.amazonaws.services.s3.model.AmazonS3Exception: Forbidden (Service: Amazon S3; Status Code: 403; Error Code: 403 Forbidden; Request ID: QKBF6K7H0D1; S3 Extended Request ID: +pWzTD5f6hHSWmN39T9izjZ4888VsMQQSoMwm9B/fjBicsfKc=; Proxy: null), S3 Extended Request ID: +pWzTD5f6hHSWmN39T9izjZ4888VsMQQKc=:403 Forbidden
	at org.apache.hadoop.fs.s3a.S3AUtils.translateException(S3AUtils.java:246)
	at org.apache.hadoop.fs.s3a.S3AUtils.translateException(S3AUtils.java:171)
	at org.apache.hadoop.fs.s3a.S3AFileSystem.s3GetFileStatus(S3AFileSystem.java:3645)
	at org.apache.hadoop.fs.s3a.S3AFileSystem.innerGetFileStatus(S3AFileSystem.java:3551)
	at org.apache.hadoop.fs.s3a.S3AFileSystem.lambda$exists$33(S3AFileSystem.java:4530)
	at org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.invokeTrackingDuration(IOStatisticsBinding.java:547)
	at org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.lambda$trackDurationOfOperation$5(IOStatisticsBinding.java:528)
	at org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.trackDuration(IOStatisticsBinding.java:449)
	at org.apache.hadoop.fs.s3a.S3AFileSystem.trackDurationAndSpan(S3AFileSystem.java:2401)
	at org.apache.hadoop.fs.s3a.S3AFileSystem.trackDurationAndSpan(S3AFileSystem.java:2420)
	at org.apache.hadoop.fs.s3a.S3AFileSystem.exists(S3AFileSystem.java:4528)
	at org.apache.hudi.util.StreamerUtil.getTableConfig(StreamerUtil.java:295)
Based on documentation https://hudi.apache.org/docs/s3_hoodie/, only explicitly defining s3 creds in core-site.xml is correct way. I would like to avoid using creds if possible, is there an option to handle it this way? Has anyone used org.apache.hadoop.fs.s3a.TemporaryAWSCredentialsProvider or
org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider
From the looks of it , it only only ssems to be happily working
org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider
org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider: this one i m not clear about Any help would be greatly appreciated.
m
Are you making sure that you use S3 Presto for checkpointing, and S3 Hadoop for reading/writing to filesystem?
n
I have used org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider
c
Are you making sure that you use S3 Presto for checkpointing, and S3 Hadoop for reading/writing to filesystem?
How do i confirm this martin? i m currently s3://. for checkpointing/savepointing and s3a://. for any activity which involves parquet files ->s3
I have used org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider
Neha, if you dont mind, could you tell me what fields have you added in your core-site.xml
Copy code
flinkConfiguration:
  classloader.parent-first-patterns.additional: org.rocksdb.
  classloader.resolve-order: parent-first
  cluster.evenly-spread-out-slots: "true"
  containerized.master.env.SINK_FILE_PATH: <s3a://flink-clen-datasets/datasets/results1>
  containerized.master.env.SINK_FILE_PATH_RCA: <s3a://flink-clen-datasets/datasets/results2>
  containerized.master.env.SINK_PARALLELISM: "4"
  execution.attached: "true"
  execution.checkpointing.interval: 5m
  execution.checkpointing.max-concurrent-checkpoints: "1"
  execution.checkpointing.min-pause: 5m
  execution.checkpointing.mode: EXACTLY_ONCE
  execution.checkpointing.snapshot-compression: "true"
  execution.checkpointing.timeout: 5m
  execution.checkpointing.unaligned: "true"
  execution.savepoint-restore-mode: CLAIM
  execution.savepoint.ignore-unclaimed-state: "true"
  fs.s3a.aws.credentials.provider: com.amazonaws.auth.WebIdentityTokenCredentialsProvider
  high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
  high-availability.storageDir: <s3://flink-clen/deployments/test-piepline1/ha>
  metrics.reporter.prom.factory.class: org.apache.flink.metrics.prometheus.PrometheusReporterFactory
  metrics.reporter.prom.port: "9250"
  metrics.reporters: prom
  rest.flamegraph.enabled: "true"
  state.backend: rocksdb
  state.backend.incremental: "true"
  state.backend.local-recovery: "true"
  state.backend.rocksdb.checkpoint.transfer.thread.num: "4"
  state.backend.rocksdb.localdir: /opt/flink/rocksdb
  state.backend.rocksdb.log.dir: /opt/flink/log/
  state.backend.rocksdb.log.level: DEBUG_LEVEL
  state.backend.rocksdb.metrics.background-errors: "true"
  state.backend.rocksdb.metrics.bytes-read: "true"
  state.backend.rocksdb.metrics.bytes-written: "true"
  state.backend.rocksdb.metrics.estimate-num-keys: "true"
  state.backend.rocksdb.metrics.is-write-stopped: "true"
  state.checkpoints.dir: <s3://flink-clen/deployments/test-piepline1/checkpoints>
  state.checkpoints.num-retained: "2"
  state.savepoints.dir: <s3://flink-clen/deployments/test-piepline1/savepoints>
  taskmanager.memory.managed.fraction: "0.2"
  taskmanager.numberOfTaskSlots: "2"
  web.cancel.enable: "false"
this is my flink config.
Copy code
flinkVersion: v1_17
serviceAccount: flink
Copy code
apiVersion: v1
kind: ServiceAccount
metadata:
  annotations:
    <http://eks.amazonaws.com/role-arn|eks.amazonaws.com/role-arn>: arn:aws:iam:::role/role
  labels:
    <http://app.kubernetes.io/name|app.kubernetes.io/name>: flink-kubernetes-operator
    <http://app.kubernetes.io/version|app.kubernetes.io/version>: 1.5.0
    <http://helm.sh/chart|helm.sh/chart>: flink-kubernetes-operator-1.5.0
  name: flink
trust relation
Copy code
"Action": "sts:AssumeRoleWithWebIdentity",
            "Condition": {
                "StringEquals": {
                    "<http://oidc.eks.us-east-1.amazonaws.com/id/:sub|oidc.eks.us-east-1.amazonaws.com/id/:sub>": "system:serviceaccount:{ns}:flink"
                }
i have covered this, this seem to be good.
adding my core-site.xml
Copy code
<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!--
  Licensed under the Apache License, Version 2.0 (the "License");
  you may not use this file except in compliance with the License.
  You may obtain a copy of the License at

    <http://www.apache.org/licenses/LICENSE-2.0>

  Unless required by applicable law or agreed to in writing, software
  distributed under the License is distributed on an "AS IS" BASIS,
  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  See the License for the specific language governing permissions and
  limitations under the License. See accompanying LICENSE file.
-->

<!-- Put site-specific property overrides in this file. -->

<configuration>
<property>
  <name>fs.s3a.assumed.role.arn</name>
  <value>arn:aws:iam:::role/role</value>
  <description>
    AWS ARN for the role to be assumed.
    Required if the fs.s3a.aws.credentials.provider contains
    org.apache.hadoop.fs.s3a.AssumedRoleCredentialProvider
  </description>
</property>

<property>
  <name>fs.s3a.assumed.role.session.name</name>
  <value>s3-connect</value>
  <description>
    Session name for the assumed role, must be valid characters according to
    the AWS APIs.
    Only used if AssumedRoleCredentialProvider is the AWS credential provider.
    If not set, one is generated from the current Hadoop/Kerberos username.
  </description>
</property>

<property>
  <name>fs.s3a.assumed.role.session.duration</name>
  <value>1h</value>
  <description>
    Duration of assumed roles before a refresh is attempted.
    Only used if AssumedRoleCredentialProvider is the AWS credential provider.
    Range: 15m to 1h
  </description>
</property>

<property>
  <name>fs.s3a.assumed.role.credentials.provider</name>
  <value>org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider</value>
  <description>
    List of credential providers to authenticate with the STS endpoint and
    retrieve short-lived role credentials.
    Used by AssumedRoleCredentialProvider and the S3A Session Delegation Token
    and S3A Role Delegation Token bindings.
  </description>
</property>
</configuration>
jars that i have added
Copy code
flink-hadoop-compatibility_2.12
hadoop-common
hadoop-aws
parquet-hadoop
aws-java-sdk-s3
hudi-flink1.16-bundle
created plugins for
plugins/s3-fs-hadoop
plugins/azure-fs-hadoop
As a pointer to anyone looking for any such setup this is all you need
Copy code
<configuration>
    <property>
        <name>fs.s3a.impl</name>
        <value>org.apache.hadoop.fs.s3a.S3AFileSystem</value>
    </property>

    <property>
        <name>fs.s3a.access.key</name>
        <value>YOUR_AWS_ACCESS_KEY</value>
    </property>

    <property>
        <name>fs.s3a.secret.key</name>
        <value>YOUR_AWS_SECRET_KEY</value>
    </property>

    <!-- Optionally, set other S3 configurations, e.g., region, endpoint, etc. -->
    <property>
        <name>fs.s3a.region</name>
        <value>us-east-1</value>
    </property>

    <!-- Set the buffer directory for S3A -->
    <property>
        <name>fs.s3a.buffer.dir</name>
        <value>/tmp</value>
    </property>
</configuration>