[destination-iceberg] support OAuth2 scope parameter in Iceberg REST catalog connection

[MChamberlin]

Connector Name

destination-iceberg

Connector Version

0.2.3

What step the error happened?

Configuring a new connector

Relevant information

It is often necessary to provide an OAuth2 scope along with client credentials for an Iceberg REST Catalog connection, particularly for the Apache Polaris REST catalog. Currently, scope is not a supported parameter for the REST catalog connection configuration, resulting in errors like this when the default scope (PRINCIPAL:ALL) is not granted to the principal.

Could not connect to the Iceberg catalog with the provided configuration. Malformed request: invalid_scope: The scope is invalid, root cause: BadRequestException(Malformed request: invalid_scope: The scope is invalid)

The Iceberg destination connector should add a scope parameter to the REST catalog configuration to support this common connection pattern.

Relevant log output

Unsetting empty environment variable 'DATA_PLANE_SERVICE_ACCOUNT_CREDENTIALS_PATH'
Unsetting empty environment variable 'KEYCLOAK_INTERNAL_REALM_ISSUER'
Unsetting empty environment variable 'DATA_PLANE_SERVICE_ACCOUNT_EMAIL'
Unsetting empty environment variable 'KEYCLOAK_CLIENT_ID'
Unsetting empty environment variable 'AIRBYTE_API_AUTH_HEADER_NAME'
Unsetting empty environment variable 'CONTROL_PLANE_AUTH_ENDPOINT'
Unsetting empty environment variable 'S3_PATH_STYLE_ACCESS'
Unsetting empty environment variable 'AIRBYTE_API_AUTH_HEADER_VALUE'
2024-12-23 01:13:38,838 [main]    INFO    i.a.c.ApplicationKt(main):12 - Sidecar start
2024-12-23 01:13:39,266 [main]    INFO    i.m.c.e.DefaultEnvironment(<init>):168 - Established active environments: [worker-v2, control-plane, oss, local-secrets]
2024-12-23 01:13:41,010 [pool-2-thread-1]    INFO    i.a.c.i.LineGobbler(voidCall):177 - 
2024-12-23 01:13:41,067 [pool-2-thread-1]    INFO    i.a.c.i.LineGobbler(voidCall):177 - ----- START CHECK -----
2024-12-23 01:13:41,068 [pool-2-thread-1]    INFO    i.a.c.i.LineGobbler(voidCall):177 - 
2024-12-23 01:13:46,622 [main]    INFO    i.a.c.ConnectorWatcher(processConnectorOutput):116 - Connector exited, processing output
2024-12-23 01:13:46,623 [main]    INFO    i.a.c.ConnectorWatcher(getConnectorOutputStream):153 - Output file jobOutput.json found
2024-12-23 01:13:46,625 [main]    INFO    i.a.c.ConnectorWatcher(processConnectorOutput):119 - Connector exited with exit code 0
2024-12-23 01:13:46,628 [main]    INFO    i.a.w.i.VersionedAirbyteStreamFactory(create):195 - Reading messages from protocol version 0.2.0
2024-12-23 01:13:46,629 [main]    WARN    i.a.m.l.MetricClientFactory(getMetricClient):43 - MetricClient has not been initialized. Must call MetricClientFactory.CreateMetricClient before using MetricClient. Using a dummy client for now. Ignore this if Airbyte is configured to not publish any metrics.
2024-12-23 01:13:46,682 [main]    INFO    i.a.w.i.VersionedAirbyteStreamFactory(internalLog):314 - INFO main i.a.c.i.b.a.AdaptiveDestinationRunner$Runner(getDestination):54 Running destination under deployment mode: OSS
2024-12-23 01:13:46,683 [main]    INFO    i.a.w.i.VersionedAirbyteStreamFactory(internalLog):314 - INFO main i.a.c.i.b.a.AdaptiveDestinationRunner$Runner(run):67 Starting destination: io.airbyte.integrations.destination.iceberg.IcebergOssDestination
2024-12-23 01:13:46,684 [main]    INFO    i.a.w.i.VersionedAirbyteStreamFactory(internalLog):314 - INFO main i.a.c.i.b.IntegrationCliParser$Companion(parseOptions):144 integration args: {check=null, config=/config/connectionConfiguration.json}
2024-12-23 01:13:46,684 [main]    INFO    i.a.w.i.VersionedAirbyteStreamFactory(internalLog):314 - INFO main i.a.c.i.b.IntegrationRunner(runInternal):130 Running integration: io.airbyte.integrations.destination.iceberg.IcebergOssDestination
2024-12-23 01:13:46,684 [main]    INFO    i.a.w.i.VersionedAirbyteStreamFactory(internalLog):314 - INFO main i.a.c.i.b.IntegrationRunner(runInternal):131 Command: CHECK
2024-12-23 01:13:46,684 [main]    INFO    i.a.w.i.VersionedAirbyteStreamFactory(internalLog):314 - INFO main i.a.c.i.b.IntegrationRunner(runInternal):132 Integration config: IntegrationConfig{command=CHECK, configPath='/config/connectionConfiguration.json', catalogPath='null', statePath='null'}
2024-12-23 01:13:46,685 [main]    WARN    i.a.w.i.VersionedAirbyteStreamFactory(internalLog):311 - WARN main c.n.s.JsonMetaSchema(newValidator):278 Unknown keyword order - you should define your own Meta Schema. If the keyword is irrelevant for validation, just use a NonValidationKeyword
2024-12-23 01:13:46,685 [main]    WARN    i.a.w.i.VersionedAirbyteStreamFactory(internalLog):311 - WARN main c.n.s.JsonMetaSchema(newValidator):278 Unknown keyword airbyte_secret - you should define your own Meta Schema. If the keyword is irrelevant for validation, just use a NonValidationKeyword
2024-12-23 01:13:46,686 [main]    ERROR    i.a.w.i.VersionedAirbyteStreamFactory(internalLog):310 - ERROR main i.a.i.d.i.BaseIcebergDestination(check):49 Exception attempting to access the Iceberg catalog: org.apache.iceberg.exceptions.BadRequestException: Malformed request: invalid_scope: The scope is invalid
    at org.apache.iceberg.rest.ErrorHandlers$OAuthErrorHandler.accept(ErrorHandlers.java:254) ~[iceberg-spark-runtime-3.5_2.13-1.5.2.jar:?]
    at org.apache.iceberg.rest.ErrorHandlers$OAuthErrorHandler.accept(ErrorHandlers.java:228) ~[iceberg-spark-runtime-3.5_2.13-1.5.2.jar:?]
    at org.apache.iceberg.rest.HTTPClient.throwFailure(HTTPClient.java:201) ~[iceberg-spark-runtime-3.5_2.13-1.5.2.jar:?]
    at org.apache.iceberg.rest.HTTPClient.execute(HTTPClient.java:313) ~[iceberg-spark-runtime-3.5_2.13-1.5.2.jar:?]
    at org.apache.iceberg.rest.HTTPClient.execute(HTTPClient.java:252) ~[iceberg-spark-runtime-3.5_2.13-1.5.2.jar:?]
    at org.apache.iceberg.rest.HTTPClient.postForm(HTTPClient.java:399) ~[iceberg-spark-runtime-3.5_2.13-1.5.2.jar:?]
    at org.apache.iceberg.rest.auth.OAuth2Util.fetchToken(OAuth2Util.java:216) ~[iceberg-spark-runtime-3.5_2.13-1.5.2.jar:?]
    at org.apache.iceberg.rest.RESTSessionCatalog.initialize(RESTSessionCatalog.java:189) ~[iceberg-spark-runtime-3.5_2.13-1.5.2.jar:?]
    at org.apache.iceberg.rest.RESTCatalog.initialize(RESTCatalog.java:78) ~[iceberg-spark-runtime-3.5_2.13-1.5.2.jar:?]
    at io.airbyte.integrations.destination.iceberg.config.catalog.RESTCatalogConfig.genCatalog(RESTCatalogConfig.java:79) ~[io.airbyte.airbyte-integrations.connectors-destination-iceberg.jar:?]
    at io.airbyte.integrations.destination.iceberg.config.catalog.IcebergCatalogConfig.check(IcebergCatalogConfig.java:43) ~[io.airbyte.airbyte-integrations.connectors-destination-iceberg.jar:?]
    at io.airbyte.integrations.destination.iceberg.BaseIcebergDestination.check(BaseIcebergDestination.java:44) [io.airbyte.airbyte-integrations.connectors-destination-iceberg.jar:?]
    at io.airbyte.cdk.integrations.base.IntegrationRunner.runInternal(IntegrationRunner.kt:166) [airbyte-cdk-core-0.44.23.jar:?]
    at io.airbyte.cdk.integrations.base.IntegrationRunner.run(IntegrationRunner.kt:119) [airbyte-cdk-core-0.44.23.jar:?]
    at io.airbyte.cdk.integrations.base.IntegrationRunner.run$default(IntegrationRunner.kt:113) [airbyte-cdk-core-0.44.23.jar:?]
    at io.airbyte.cdk.integrations.base.adaptive.AdaptiveDestinationRunner$Runner.run(AdaptiveDestinationRunner.kt:68) [airbyte-cdk-core-0.44.23.jar:?]
    at io.airbyte.integrations.destination.iceberg.IcebergDestinationRunner.main(IcebergDestinationRunner.java:15) [io.airbyte.airbyte-integrations.connectors-destination-iceberg.jar:?]

Stack Trace: org.apache.iceberg.exceptions.BadRequestException: Malformed request: invalid_scope: The scope is invalid
    at org.apache.iceberg.rest.ErrorHandlers$OAuthErrorHandler.accept(ErrorHandlers.java:254)
    at org.apache.iceberg.rest.ErrorHandlers$OAuthErrorHandler.accept(ErrorHandlers.java:228)
    at org.apache.iceberg.rest.HTTPClient.throwFailure(HTTPClient.java:201)
    at org.apache.iceberg.rest.HTTPClient.execute(HTTPClient.java:313)
    at org.apache.iceberg.rest.HTTPClient.execute(HTTPClient.java:252)
    at org.apache.iceberg.rest.HTTPClient.postForm(HTTPClient.java:399)
    at org.apache.iceberg.rest.auth.OAuth2Util.fetchToken(OAuth2Util.java:216)
    at org.apache.iceberg.rest.RESTSessionCatalog.initialize(RESTSessionCatalog.java:189)
    at org.apache.iceberg.rest.RESTCatalog.initialize(RESTCatalog.java:78)
    at io.airbyte.integrations.destination.iceberg.config.catalog.RESTCatalogConfig.genCatalog(RESTCatalogConfig.java:79)
    at io.airbyte.integrations.destination.iceberg.config.catalog.IcebergCatalogConfig.check(IcebergCatalogConfig.java:43)
    at io.airbyte.integrations.destination.iceberg.BaseIcebergDestination.check(BaseIcebergDestination.java:44)
    at io.airbyte.cdk.integrations.base.IntegrationRunner.runInternal(IntegrationRunner.kt:166)
    at io.airbyte.cdk.integrations.base.IntegrationRunner.run(IntegrationRunner.kt:119)
    at io.airbyte.cdk.integrations.base.IntegrationRunner.run$default(IntegrationRunner.kt:113)
    at io.airbyte.cdk.integrations.base.adaptive.AdaptiveDestinationRunner$Runner.run(AdaptiveDestinationRunner.kt:68)
    at io.airbyte.integrations.destination.iceberg.IcebergDestinationRunner.main(IcebergDestinationRunner.java:15)

2024-12-23 01:13:46,687 [main]    INFO    i.a.w.i.VersionedAirbyteStreamFactory(internalLog):314 - INFO main i.a.c.i.b.IntegrationRunner(runInternal):224 Completed integration: io.airbyte.integrations.destination.iceberg.IcebergOssDestination
2024-12-23 01:13:46,687 [main]    INFO    i.a.w.i.VersionedAirbyteStreamFactory(internalLog):314 - INFO main i.a.c.i.b.a.AdaptiveDestinationRunner$Runner(run):69 Completed destination: io.airbyte.integrations.destination.iceberg.IcebergOssDestination
2024-12-23 01:13:46,689 [main]    INFO    i.a.c.ConnectorMessageProcessor(updateConfigFromControlMessage):232 - Checking for optional control message...
2024-12-23 01:13:46,689 [main]    INFO    i.a.c.ConnectorMessageProcessor(updateConfigFromControlMessage):254 - Optional control message not found. Skipping...
2024-12-23 01:13:46,691 [main]    INFO    i.a.c.ConnectorWatcher(saveConnectorOutput):164 - Writing output of df65a8f3-9908-451b-aa9b-445462803560_9ac480f1-03d9-422d-b1dd-8f8aa0e96bd4_0_check to the doc store
2024-12-23 01:13:46,990 [main]    INFO    i.a.c.ConnectorWatcher(markWorkloadSuccess):169 - Marking workload df65a8f3-9908-451b-aa9b-445462803560_9ac480f1-03d9-422d-b1dd-8f8aa0e96bd4_0_check as successful
2024-12-23 01:13:47,000 [pool-4-thread-1]    INFO    i.a.c.i.LineGobbler(voidCall):177 - 
2024-12-23 01:13:47,000 [main]    INFO    i.a.c.ConnectorWatcher(exitProperly):217 - Deliberately exiting process with code 0.
2024-12-23 01:13:47,000 [pool-4-thread-1]    INFO    i.a.c.i.LineGobbler(voidCall):177 - ----- END CHECK -----
2024-12-23 01:13:47,001 [pool-4-thread-1]    INFO    i.a.c.i.LineGobbler(voidCall):177 - 
Stream closed EOF for airbyte/tion-iceberg-check-9ac480f1-03d9-422d-b1dd-8f8aa0e96bd4-0-bsfsq (connector-sidecar)

Contribute

• Yes, I want to contribute

[marcosmarxm]

Hello @MChamberlin thanks for reporting the issue. Looking into the codebase it seems the change to fix the issue is something like:

• add optional parameter role to the REST Catalog in spec-oss.json file;
• then edit the RESTCatalogConfig.java to build the Spark config properly. There are some examples using optional args.

Let me know if you need any help to make the contribution.