{"record":{"id":"5fd7f8605714813d","repo":"apache/beam","slug":"pool-shard-subscriber-got-error","errorCode":null,"errorMessage":"Pool {} - shard {} subscriber got error","messagePattern":"Pool (.+?) - shard (.+?) subscriber got error","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/EFOShardSubscriber.java","lineNumber":332,"sourceCode":"      pool.enqueueEvent(shardId, event);\n      sequenceNumber = event.continuationSequenceNumber();\n      int capacity = pool.getMaxCapacityPerShard() - inFlight.incrementAndGet();\n      checkState(capacity >= 0, \"Exceeded in-flight limit\");\n      if (capacity > 0 && subscription != null) {\n        subscription.request(1);\n      }\n    }\n\n    /** Delegates to {@link #visit}. */\n    @Override\n    public void onNext(SubscribeToShardEventStream event) {\n      event.accept(this);\n    }\n\n    /** Nothing to do here, handled in {@link #reSubscriptionHandler}. */\n    @Override\n    public void onError(Throwable t) {\n      LOG.warn(\"Pool {} - shard {} subscriber got error\", pool.getPoolId(), shardId, t);\n    }\n\n    /** Unsets {@link #eventsSubscriber} of {@link EFOShardSubscriber}. */\n    @Override\n    public void onComplete() {\n      subscription = null;\n    }\n  }\n}\n","sourceCodeStart":314,"sourceCodeEnd":342,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/EFOShardSubscriber.java#L314-L342","documentation":"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.","triggerScenarios":"The Kinesis SubscribeToShardEventStream publisher emits an error (connection drop, access denied, expired shard subscription, Kinesis throttling) and the subscriber's onError callback fires.","commonSituations":"Enhanced fan-out consumer losing its stream connection during network blips; IAM credentials expiring mid-stream; shard subscription limits reached; Kinesis service throttling.","solutions":["Verify the reSubscriptionHandler is wired with a retry policy (backoff) — recovery is automatic.","Check IAM permissions for the EFO consumer (SubscribeToShard).","Inspect the logged throwable; if AccessDenied/ResourceNotFound recurs, fix credentials/consumer ARN configuration.","Confirm the EFO consumer is registered and stream/shard exists."],"exampleFix":"// before\n.onError(t -> LOG.warn(\"Pool {} - shard {} subscriber got error\", pool.getPoolId(), shardId, t));\n// after\n// ensure retry in reSubscriptionHandler:\nRetry.backoff(Long.MAX_VALUE, Duration.ofSeconds(1))\n    .doOnNext(s -> LOG.info(\"Re-subscribing shard {}\", shardId))\n    .subscribe(...);","handlingStrategy":"retry","validationCode":"// Pre-flight EFO consumer registration & permissions:\nawsCli kinesis describe-stream-consumer --stream-arn ... --consumer-name ...","typeGuard":null,"tryCatchPattern":"efoSubscriber\n  .onError(t -> {\n    LOG.warn(\"subscriber got error\", t);\n    // rely on reSubscriptionHandler retry/backoff\n  });","preventionTips":["Configure exponential backoff in the re-subscription handler","Ensure the EFO consumer is registered and IAM allows SubscribeToShard","Refresh credentials before expiry in long-running consumers","Monitor logs for repeated onError on the same shard"],"tags":["aws","kinesis","streaming","resilience"],"backgroundTag":"http-error-response","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T21:17:11.552Z"}