{"record":{"id":"2af85835d2be3624","repo":"apache/beam","slug":"failed-to-refresh-shards","errorCode":null,"errorMessage":"Failed to refresh shards.","messagePattern":"Failed to refresh shards\\.","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/KinesisIO.java","lineNumber":1261,"sourceCode":"        @SuppressWarnings(\"FutureReturnValueIgnored\") // safe to ignore\n        private void refresh(\n            KinesisAsyncClient client,\n            Supplier<Instant> nextRefreshFn,\n            NavigableSet<BigInteger> bounds,\n            @Nullable String nextToken) {\n          ListShardsRequest.Builder reqBuilder =\n              ListShardsRequest.builder().shardFilter(f -> f.type(AT_LATEST));\n          if (nextToken != null) {\n            reqBuilder.nextToken(nextToken);\n          } else {\n            reqBuilder.streamName(streamName);\n          }\n          client\n              .listShards(reqBuilder.build())\n              .whenComplete(\n                  (resp, exc) -> {\n                    if (exc != null) {\n                      LOG.warn(\"Failed to refresh shards.\", exc);\n                      nextRefresh = nextRefreshFn.get(); // retry later\n                      running.set(false);\n                      return;\n                    }\n                    resp.shards().forEach(shard -> bounds.add(lowerHashKey(shard)));\n                    if (resp.nextToken() != null) {\n                      refresh(client, nextRefreshFn, bounds, resp.nextToken());\n                      return;\n                    }\n                    LOG.debug(\"Done refreshing {} shards.\", bounds.size());\n                    nextRefresh = nextRefreshFn.get();\n                    running.set(false);\n                    shardBounds = bounds; // swap key ranges\n                  });\n        }\n\n        private BigInteger lowerHashKey(Shard shard) {\n          return new BigInteger(shard.hashKeyRange().startingHashKey());","sourceCodeStart":1243,"sourceCodeEnd":1279,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/KinesisIO.java#L1243-L1279","documentation":"The periodic background refresh of Kinesis shard boundaries (via listShards, asynchronously) failed. The refresh task logs the exception, resets the running flag, and schedules a retry at the next refresh interval; writes continue with the previously cached shard bounds.","triggerScenarios":"client.listShards(...) async call completed exceptionally — expired/invalid AWS credentials, throttling (LimitExceededException), network failure, or the producer lacks kinesis:ListShards IAM permission.","commonSituations":"IAM policies that grant PutRecord but not ListShards; credential rotation failures; Kinesis throttling on streams with many concurrent listShards callers; transient network blips in VPC endpoints.","solutions":["Grant kinesis:ListShards permission to the producer's IAM role","Check AWS credentials validity/expiry in the pipeline's AwsOptions provider","Increase withRefreshPeriod and add AWS SDK retry/timeout tuning to reduce throttling","No action needed for transient failures — the refresher retries at the next interval automatically"],"exampleFix":"// IAM policy before (missing list)\n{\"Effect\":\"Allow\",\"Action\":[\"kinesis:PutRecord*\"],\"Resource\":\"*\"}\n// after\n{\"Effect\":\"Allow\",\"Action\":[\"kinesis:PutRecord*\",\"kinesis:ListShards\"],\"Resource\":\"*\"}","handlingStrategy":"retry","validationCode":"// preflight IAM check\ntry { client.listShards(b -> b.streamName(stream).maxResults(1)); }\ncatch (Exception e) { throw new IllegalStateException(\"ListShards not permitted/failing: \" + e.getMessage()); }","typeGuard":null,"tryCatchPattern":"// the library already retries on next refresh; for your own calls:\nCompletableFuture<ListShardsResponse> f = client.listShards(req);\nf.exceptionally(exc -> { log.warn(\"shard refresh failed, keeping stale bounds\", exc); return lastGoodResponse; });","preventionTips":["Grant kinesis:ListShards to the pipeline role","Use a credentials provider that auto-refreshes before expiry","Set a refresh period comfortably above Kinesis rate limits","Alert on this warning recurring continuously"],"tags":["kinesis","aws","iam","throttling","background-refresh"],"backgroundTag":"api-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"}