apache/beam · error · IllegalArgumentException

Service endpoint must be a URI, got

Error message

Service endpoint must be a URI, got: %s

What it means

KinesisWriteSchemaTransformProvider.expand parses config.getServiceEndpoint() with new URI(...); a URISyntaxException is wrapped as an IllegalArgumentException including the bad value. Applies to the Kinesis write schema transform (YAML pipeline).

Solutions

1. Provide a full URI: https://kinesis.region.amazonaws.com or http://localhost:4566 for localstack
2. Pre-validate with new URI(value) or a regex for scheme://host:port
3. Unset service_endpoint to use the AWS default

Example fix

// before
service_endpoint: "${KINESIS_ENDPOINT}"
// after
service_endpoint: https://kinesis.us-east-1.amazonaws.com

try { new URI(config.getServiceEndpoint()); } catch (URISyntaxException e) { throw new IllegalArgumentException("bad service_endpoint", e); }

boolean validEndpoint(String s){ try { new URI(s); return true; } catch (Exception e){ return false; } }

try { expand(cfg); } catch (IllegalArgumentException e) { fix endpoint scheme/syntax in YAML; resubmit; }

When it happens

Trigger: service_endpoint in a Kinesis write YAML config that is not valid URI syntax (no scheme, illegal chars, spaces).

Common situations: Missing https://; localstack endpoint without scheme; templated env var left unresolved like ${KINESIS_ENDPOINT}.

Understand the failure class

Background: "Invalid URL" errors: why new URL(), URI.parse, and reqwest::Url reject your string — missing scheme, whitespace, and bad path format — this error's family across 39 libraries.

Related errors

• Service endpoint must be a URI, got
• Service endpoint must be URI format, got
• Expected input schema with a 'data' (BYTES) field, a…
• Unsupported watermark_policy:
• Wrong JDBC type. Expected

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/be431f4be08249f6. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/KinesisWriteSchemaTransformProvider.java:183

    public PCollectionRowTuple expand(PCollectionRowTuple input) {
      PCollection<Row> inputRows = input.getSinglePCollection();
      Schema schema = inputRows.getSchema();

      final int dataFieldIndex = resolveDataFieldIndex(schema);
      final boolean dataIsString = isStringField(schema, dataFieldIndex);
      final Integer partitionKeyIndex =
          schema.hasField("partition_key") ? schema.indexOf("partition_key") : null;

      AwsBasicCredentials creds =
          AwsBasicCredentials.create(config.getAwsAccessKey(), config.getAwsSecretKey());
      StaticCredentialsProvider provider = StaticCredentialsProvider.create(creds);

      @Nullable URI endpoint = null;
      if (config.getServiceEndpoint() != null) {
        try {
          endpoint = new URI(config.getServiceEndpoint());
        } catch (URISyntaxException ex) {
          throw new IllegalArgumentException(
              String.format("Service endpoint must be a URI, got: %s", config.getServiceEndpoint()),
              ex);
        }
      }

      boolean verifyCertificate =
          config.getVerifyCertificate() == null || config.getVerifyCertificate();

      SerializableFunction<KV<String, byte[]>, byte[]> serializer = KV::getValue;
      KinesisIO.Write<KV<String, byte[]>> writeTransform =
          KinesisIO.<KV<String, byte[]>>write()
              .withStreamName(config.getStreamName())
              .withClientConfiguration(
                  ClientConfiguration.builder()
                      .credentialsProvider(provider)
                      .region(Region.of(config.getRegion()))
                      .endpoint(endpoint)
                      .skipCertificateVerification(!verifyCertificate)

View on GitHub (pinned to 12126d8942)