apache/beam · warning

Pool {} - shard {} subscriber got error

Error message

Pool {} - shard {} subscriber got error

What it means

EFOShardSubscriber's reactive subscription received onError from the upstream Kinesis enhanced fan-out event stream. The error is only logged; recovery is handled by reSubscriptionHandler, which re-subscribes the shard. This warning is expected during transient Kinesis/Network disruptions.

Source

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

      pool.enqueueEvent(shardId, event);
      sequenceNumber = event.continuationSequenceNumber();
      int capacity = pool.getMaxCapacityPerShard() - inFlight.incrementAndGet();
      checkState(capacity >= 0, "Exceeded in-flight limit");
      if (capacity > 0 && subscription != null) {
        subscription.request(1);
      }
    }

    /** Delegates to {@link #visit}. */
    @Override
    public void onNext(SubscribeToShardEventStream event) {
      event.accept(this);
    }

    /** Nothing to do here, handled in {@link #reSubscriptionHandler}. */
    @Override
    public void onError(Throwable t) {
      LOG.warn("Pool {} - shard {} subscriber got error", pool.getPoolId(), shardId, t);
    }

    /** Unsets {@link #eventsSubscriber} of {@link EFOShardSubscriber}. */
    @Override
    public void onComplete() {
      subscription = null;
    }
  }
}

View on GitHub (pinned to 12126d8942)

Solutions

1. Verify the reSubscriptionHandler is wired with a retry policy (backoff) — recovery is automatic.
2. Check IAM permissions for the EFO consumer (SubscribeToShard).
3. Inspect the logged throwable; if AccessDenied/ResourceNotFound recurs, fix credentials/consumer ARN configuration.
4. Confirm the EFO consumer is registered and stream/shard exists.

Example fix

// before
.onError(t -> LOG.warn("Pool {} - shard {} subscriber got error", pool.getPoolId(), shardId, t));
// after
// ensure retry in reSubscriptionHandler:
Retry.backoff(Long.MAX_VALUE, Duration.ofSeconds(1))
    .doOnNext(s -> LOG.info("Re-subscribing shard {}", shardId))
    .subscribe(...);

// Pre-flight EFO consumer registration & permissions:
awsCli kinesis describe-stream-consumer --stream-arn ... --consumer-name ...

efoSubscriber
  .onError(t -> {
    LOG.warn("subscriber got error", t);
    // rely on reSubscriptionHandler retry/backoff
  });

When it happens

Trigger: The Kinesis SubscribeToShardEventStream publisher emits an error (connection drop, access denied, expired shard subscription, Kinesis throttling) and the subscriber's onError callback fires.

Common situations: Enhanced fan-out consumer losing its stream connection during network blips; IAM credentials expiring mid-stream; shard subscription limits reached; Kinesis service throttling.

Understand the failure class

Background: "API error: {status}" and "HTTP 401/403/404/429/5xx" errors: non-2xx HTTP responses explained — this error's family across 27 libraries.

Related errors

• Kinesis backend failed. Wait some time and retry.
• Thread was interrupted, finishing the read loop
• Transient exception occurred.
• Interrupted while waiting for KinesisRecord from the buffer
• Failed to refresh shards.

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