apache/beam · error · RuntimeException

Service endpoint must be URI format, got

Error message

Service endpoint must be URI format, got: %s

What it means

KinesisTransformRegistrar.buildExternal (write path) parses configuration.serviceEndpoint into a java.net.URI. On URISyntaxException it throws a RuntimeException with a message naming the offending value. The endpoint, when present, must be valid URI syntax.

Solutions

1. Prefix the endpoint with a scheme (https:// or http:// for localstack/Kinesalite)
2. Validate with new URI(endpoint) before submitting the pipeline JSON
3. Drop the serviceEndpoint field to use the default AWS endpoint

Example fix

// before
"serviceEndpoint": "kinesis.eu-west-1.amazonaws.com"
// after
"serviceEndpoint": "https://kinesis.eu-west-1.amazonaws.com"

try { new URI(configuration.serviceEndpoint); } catch (URISyntaxException e) { /* fix before submission */ }

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

try { registrar.buildExternal(cfg); } catch (RuntimeException e) { correct serviceEndpoint in the pipeline JSON; retry expansion; }

When it happens

Trigger: Cross-language (transform service) pipeline with a kinesis write transform whose serviceEndpoint config is malformed (missing scheme, illegal characters).

Common situations: Hand-authored JSON pipeline options; missing https:// prefix; port typos like 'host::8080'; unexpanded template variables.

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 a URI, got
• Unsupported watermark_policy:
• Unsupported watermark policy type
• Allow list file does not exist

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

Appendix: source

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

      public void setPartitionKey(String partitionKey) {
        this.partitionKey = partitionKey;
      }
    }

    @Override
    public PTransform<PCollection<byte[]>, KinesisIO.Write.Result> buildExternal(
        Configuration configuration) {
      AwsBasicCredentials creds =
          AwsBasicCredentials.create(configuration.awsAccessKey, configuration.awsSecretKey);
      String pk = configuration.partitionKey;
      StaticCredentialsProvider provider = StaticCredentialsProvider.create(creds);
      SerializableFunction<byte[], byte[]> serializer = v -> v;
      @Nullable URI endpoint = null;
      if (configuration.serviceEndpoint != null) {
        try {
          endpoint = new URI(configuration.serviceEndpoint);
        } catch (URISyntaxException ex) {
          throw new RuntimeException(
              String.format(
                  "Service endpoint must be URI format, got: %s", configuration.serviceEndpoint));
        }
      }
      KinesisIO.Write<byte[]> writeTransform =
          KinesisIO.<byte[]>write()
              .withStreamName(configuration.streamName)
              .withClientConfiguration(
                  ClientConfiguration.builder()
                      .credentialsProvider(provider)
                      .region(Region.of(configuration.region))
                      .endpoint(endpoint)
                      .skipCertificateVerification(!configuration.verifyCertificate)
                      .build())
              .withPartitioner(p -> pk)
              .withSerializer(serializer);

      if (configuration.aggregationEnabled) {

View on GitHub (pinned to 12126d8942)