Package io.confluent.flink.plugin

Class ConfluentSettings.Builder

java.lang.Object
io.confluent.flink.plugin.ConfluentSettings.Builder
Enclosing class:
ConfluentSettings
public static class ConfluentSettings.Builder extends Object
Builder pattern for a fluent definition of Confluent-specific EnvironmentSettings.

  • Method Details

    • applyArgs

      public ConfluentSettings.Builder applyArgs(String[] args)
      Applies configuration from CLI arguments on top of the current builder state.

      Unlike ConfluentSettings.newBuilderFromArgs(String[]), which only applies the ConfluentPluginOptions.VAR_FLINK_PROPERTIES environment variable as a fallback before parsing args, this method lets a program layer arguments onto any existing builder. A program can therefore combine a base configuration (for example loaded via ConfluentSettings.newBuilderFromResource(String), ConfluentSettings.newBuilderFromFile(File), or explicit setters) with per-invocation arguments: 

       EnvironmentSettings settings =
     ConfluentSettings.newBuilder()              // FLINK_PROPERTIES file + env-var fallback
         .setApplicationName("my-table-program") // code default, overridable by --application-name
         .applyArgs(args)                        // per-invocation overrides + actions
         .build();

      Only the keys present in args are written; absent keys leave any previously configured value untouched. Because parsing writes into the same Configuration as the builder setters, arguments follow last-writer-wins semantics: values applied here override earlier builder state, and later builder calls override these.

      The parsed options are those listed in ConfluentPluginOptions.ARG_OPTIONS. A PluginAction passed as the leading argument (for example stop) is recorded and dispatched when build() is called.

      Parameters:
      args - the main(String[] args) arguments
      Returns:
      this builder

    • setApplicationName

      public ConfluentSettings.Builder setApplicationName(String name)
      Sets a name for this Table API application. Optional but recommended.

      The application name serves as a namespace for all statements submitted by this application. It will be used as a prefix for all statement names, whether auto-generated or explicitly set via ConfluentTools.setStatementName(TableEnvironment, String).

      Behavior:

      • Auto-generated statement names: "{applicationName}-{uuid}"
      • Explicit statement names: "{applicationName}-{statementName}"

      If you do not set an application name:

      • Auto-generated statement names will use: "table-api-{date}-{uuid}"
      • Explicit statement names will be used as-is (fully qualified)

      Naming constraints:

      • Must contain only lowercase alphanumeric characters and hyphens
      • Must start and end with an alphanumeric character (not a hyphen)
      • Because it prefixes statement names, the effective usable length must be smaller to keep the full statement name within the 100-character limit.

      For example: "my-table-program".

      Parameters:
      name - the application name
      Returns:
      this builder

    • setOrganizationId

      public ConfluentSettings.Builder setOrganizationId(String orgId)
      Sets the ID of the organization. Required.

      For example: 'b0b21724-4586-4a07-b787-d0bb5aacbf87'.

      Overrides the global variable ORG_ID.

    • setEnvironmentId

      public ConfluentSettings.Builder setEnvironmentId(String envId)
      Sets the ID of the environment. Required.

      For example: 'env-z3y2x1'.

      Overrides the global variable ENV_ID.

    • setFlinkApiKey

      public ConfluentSettings.Builder setFlinkApiKey(String key)
      Sets the API key for Flink access. Required.

      Overrides global variable FLINK_API_KEY.

    • setFlinkApiSecret

      public ConfluentSettings.Builder setFlinkApiSecret(String secret)
      Sets the API secret for Flink access. Required.

      Overrides global variable FLINK_API_SECRET.

    • setArtifactApiKey

      public ConfluentSettings.Builder setArtifactApiKey(String key)
      Sets the API key for Artifacts creation. Optional.

      Overrides global variable ARTIFACT_API_KEY.

    • setArtifactApiSecret

      public ConfluentSettings.Builder setArtifactApiSecret(String secret)
      Sets the API secret for Artifacts creation. Optional.

      Overrides global variable ARTIFACT_API_SECRET.

    • setGlobalApiKey

      public ConfluentSettings.Builder setGlobalApiKey(String key)
      Sets the global API key used for both Flink and Artifact access. Optional.

      When set together with setGlobalApiSecret(java.lang.String), it takes precedence over the dedicated Flink and Artifact API keys.

      Overrides global variable GLOBAL_API_KEY.

    • setGlobalApiSecret

      public ConfluentSettings.Builder setGlobalApiSecret(String secret)
      Sets the global API secret used for both Flink and Artifact access. Optional.

      When set together with setGlobalApiKey(java.lang.String), it takes precedence over the dedicated Flink and Artifact API secrets.

      Overrides global variable GLOBAL_API_SECRET.

    • setTemporaryDirectory

      public ConfluentSettings.Builder setTemporaryDirectory(String dir)
      Sets the directory for temporary files, e.g. UDF jars. Optional.

    • setHttpUserAgent

      public ConfluentSettings.Builder setHttpUserAgent(String userAgent)
      Sets the HTTP User-Agent for API requests. Optional.

      If not set, a default User-Agent will be used based on the package name and version.

      Overrides global variable HTTP_USER_AGENT.

    • setComputePoolId

      public ConfluentSettings.Builder setComputePoolId(String computePoolId)
      Sets the ID of the compute pool. Required.

      For example: 'lfcp-8m03rm'.

      Overrides global variable COMPUTE_POOL_ID.

    • setCloud

      public ConfluentSettings.Builder setCloud(String provider)
      Sets the Confluent identifier for a cloud provider. Required.

      For example: 'aws'.

      Overrides the global variable CLOUD_PROVIDER.

    • setRegion

      public ConfluentSettings.Builder setRegion(String region)
      Sets the Confluent identifier for a cloud provider's region. Required.

      For example: 'us-east-1'.

      Overrides the global variable CLOUD_REGION.

    • setPrincipalId

      public ConfluentSettings.Builder setPrincipalId(String principalId)
      Sets the Principal that runs submitted statements. Optional.

      For example: 'sa-23kgz4' (for a service account).

      Overrides the global variable PRINCIPAL_ID.

    • setRestEndpoint

      public ConfluentSettings.Builder setRestEndpoint(String endpoint)
      Sets the URL to the REST endpoint. Optional.

      For example: 'confluent.cloud'.

      Overrides the global variable REST_ENDPOINT.

    • setEndpointTemplate

      public ConfluentSettings.Builder setEndpointTemplate(String endpointTemplate)
      Sets the template for the endpoint URL. Optional.

      For example: 'https://flinkpls-abc123.{region}.{cloud}.glb.confluent.cloud'.

      The template may contain the placeholders {region} and {cloud}. The placeholders will be replaced with the values of the variable CLOUD_REGION and CLOUD_PROVIDER respectively.

      Default value is 'https://flink.{region}.{cloud}.confluent.cloud'.

      Overrides the global variable ENDPOINT_TEMPLATE.

    • setArtifactEndpointTemplate

      public ConfluentSettings.Builder setArtifactEndpointTemplate(String artifactEndpointTemplate)
      Sets the template for the artifact endpoint URL. Optional.

      For example: 'https://api.{region}.{cloud}.confluent.cloud'.

      The template may contain the placeholders {region} and {cloud}. The placeholders will be replaced with the values of the variable CLOUD_REGION and CLOUD_PROVIDER respectively.

      Default value is 'https://api.confluent.cloud'.

      Overrides the global variable ARTIFACT_ENDPOINT_TEMPLATE.

    • setCatalogCache

      public ConfluentSettings.Builder setCatalogCache(Duration expiration)
      Sets the expiration time for catalog objects before Confluent Cloud is called again. Optional.

      For example: '5 min'. '1 min' by default. '0' disables the caching.

    • setAuthMode

      public ConfluentSettings.Builder setAuthMode(AuthMode authMode)
      Sets the authentication mode for the Confluent Cloud REST APIs. Optional, defaults to AuthMode.API_KEY.

      UDF artifact uploads additionally require either a global API key (setGlobalApiKey(java.lang.String), setGlobalApiSecret(java.lang.String)) or a dedicated Artifact API key (setArtifactApiKey(java.lang.String), setArtifactApiSecret(java.lang.String)). A global API key, when set, takes precedence over the dedicated Flink and Artifact API keys.

      Overrides the global variable AUTH_MODE.

    • setOAuthExternalTokenUrl

      public ConfluentSettings.Builder setOAuthExternalTokenUrl(String tokenUrl)
      Sets the URL of the external IdP's OAuth 2.0 token endpoint. Required when auth-mode=OAUTH_CLIENT_CREDENTIALS.

      For example: 'https://mycompany.okta.com/oauth2/abc123/v1/token'.

      Overrides the global variable OAUTH_EXTERNAL_TOKEN_URL.

    • setOAuthExternalClientId

      public ConfluentSettings.Builder setOAuthExternalClientId(String clientId)
      Sets the client ID registered with the external IdP. Required when auth-mode=OAUTH_CLIENT_CREDENTIALS.

      Overrides the global variable OAUTH_EXTERNAL_CLIENT_ID.

    • setOAuthExternalClientSecret

      public ConfluentSettings.Builder setOAuthExternalClientSecret(String clientSecret)
      Sets the client secret registered with the external IdP. Required when auth-mode=OAUTH_CLIENT_CREDENTIALS.

      Treated as sensitive: the plugin never logs this value.

      Overrides the global variable OAUTH_EXTERNAL_CLIENT_SECRET.

    • setOAuthExternalTokenScope

      public ConfluentSettings.Builder setOAuthExternalTokenScope(String scope)
      Sets the OAuth scope to request from the IdP. Optional and IdP-dependent.

      Microsoft Entra ID typically requires api://<client_id>/.default. Okta does not require any scope. Used when auth-mode=OAUTH_CLIENT_CREDENTIALS.

      Overrides the global variable OAUTH_EXTERNAL_TOKEN_SCOPE.

    • setOAuthExternalAccessToken

      public ConfluentSettings.Builder setOAuthExternalAccessToken(String accessToken)
      Sets a pre-issued OAuth bearer token. Used when auth-mode=OAUTH_STATIC_TOKEN.

      The user is responsible for ensuring the token is not expired; the plugin does not refresh it. Mutually exclusive with setOAuthTokenProvider(io.confluent.flink.plugin.OAuthTokenProvider).

      Overrides the global variable OAUTH_EXTERNAL_ACCESS_TOKEN.

    • setOAuthIdentityPoolId

      public ConfluentSettings.Builder setOAuthIdentityPoolId(String identityPoolId)
      Sets the Confluent Cloud identity pool ID. Used when auth-mode=OAUTH_STATIC_TOKEN or auth-mode=OAUTH_CLIENT_CREDENTIALS

      For example: 'pool-xxxxx'. The plugin sends this on every Flink SQL request as the Confluent-Identity-Pool-Id header.

      Overrides the global variable OAUTH_IDENTITY_POOL_ID.

    • setOAuthTokenProvider

      public ConfluentSettings.Builder setOAuthTokenProvider(OAuthTokenProvider provider)
      Registers a programmatic OAuth token provider for cloud-native identity flows (Azure Managed Identity, AWS IAM workload identity, custom token services).

      Requires setAuthMode(io.confluent.flink.plugin.AuthMode) with AuthMode.OAUTH_STATIC_TOKEN on the same builder.

    • setStatementName

      public ConfluentSettings.Builder setStatementName(String name)
      Sets the statement name for statement submission.

      A statement name must be unique within an environment and cloud region for a given organization. By default, statement names are auto-generated using a UUID.

      Important: If you configured an application name via setApplicationName(String), it will be automatically prefixed to the statement name you provide here. For example, if your application name is "myapp" and you set the statement name to "query1", the final statement name will be "myapp-query1".

      If you did not configure an application name and use this method to set an explicit statement name, the statement name you provide will be used as-is (fully qualified).

      First submission only: When set via ConfluentSettings, the statement name applies only to the first statement submitted on the resulting TableEnvironment; it is automatically reset afterward, and subsequent submissions auto-generate a unique name. To simply run multiple statements without managing names, configure an application name via setApplicationName(String) instead, which namespaces every statement's auto-generated name. When a specific statement needs a stable name to be stopped, resumed, or replaced across runs, use ConfluentTools.setStatementName(TableEnvironment, String) before each such submission.

      Naming constraints:

      • Must contain only lowercase alphanumeric characters and hyphens
      • Must start and end with an alphanumeric character (not a hyphen)
      • Maximum length: 100 characters (including the application name prefix if configured)
      Parameters:
      name - the statement name (will be prefixed with application name if configured)
      Returns:
      this builder

    • setOnConflict

      public ConfluentSettings.Builder setOnConflict(OnConflictMode mode)
      Sets the behavior when a statement could not be updated, due to a conflict. Optional, defaults to OnConflictMode.FAIL.

      With OnConflictMode.REPLACE, on a conflict during statement submission, the plugin deletes the conflicting statement and retries the submission once. Requires an application name to be configured via setApplicationName(String) so that statement names are stable across runs.

    • setOption

      public ConfluentSettings.Builder setOption(String key, String value)
      Sets a Confluent-specific configuration option.

      Note: This method does not support Flink options. Check the documentation for more information.

    • build

      public EnvironmentSettings build()
      Creates Confluent-specific EnvironmentSettings that can be passed into TableEnvironment.create(EnvironmentSettings).