{"record":{"id":"9a0c60123011b806","repo":"apache/druid","slug":"streamexception-9a0c60","errorCode":null,"errorMessage":"StreamException","messagePattern":"StreamException","errorType":"exception","errorClass":"StreamException","httpStatus":null,"severity":"error","filePath":"extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java","lineNumber":123,"sourceCode":"\n  /**\n   * We call getRecords with limit 1000 to make sure that we can find the first (earliest) record in the shard.\n   * In the case where the shard is constantly removing records that are past their retention period, it is possible\n   * that we never find the first record in the shard if we use a limit of 1.\n   */\n  private static final int GET_SEQUENCE_NUMBER_RECORD_COUNT = 1000;\n  private static final int GET_SEQUENCE_NUMBER_RETRY_COUNT = 10;\n\n  /**\n   * Catch any exception and wrap it in a {@link StreamException}\n   */\n  private static <T> T wrapExceptions(Callable<T> callable)\n  {\n    try {\n      return callable.call();\n    }\n    catch (Exception e) {\n      throw new StreamException(e);\n    }\n  }\n\n  private class PartitionResource\n  {\n    private final StreamPartition<String> streamPartition;\n\n    // shardIterator points to the record that will be polled next by recordRunnable\n    // can be null when shard is closed due to the user shard splitting or changing the number\n    // of shards in the stream, in which case a 'EOS' marker is used by the KinesisRecordSupplier\n    // to indicate that this shard has no more records to read\n    @Nullable\n    private volatile String shardIterator;\n    private volatile long currentLagMillis;\n\n    private final AtomicBoolean fetchStarted = new AtomicBoolean();\n    private ScheduledFuture<?> currentFetch;\n","sourceCodeStart":105,"sourceCodeEnd":141,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java#L105-L141","documentation":"KinesisRecordSupplier wraps every checked/underlying exception thrown by AWS SDK calls inside StreamException via the wrapExceptions helper. Any failure while talking to Kinesis (iterator expiry, throttling, credentials, network) surfaces as StreamException with the original cause attached. It exists so callers get one uniform unchecked exception type for all Kinesis I/O failures.","triggerScenarios":"Calling isOffsetAvailable, getPartitionIds, getSequenceNumber, or getPartitionTimeLag when the wrapped Kinesis call (getShardIterator, ListShards, getRecords) throws any Exception: expired iterator, AccessDeniedException, throttling, network outage, or invalid shard/sequence number.","commonSituations":"IAM role missing kinesis:Read permissions on the stream; shard iterator expired after >5 min inactivity; stream deleted or region misconfigured; transient network blips during ingestion task startup.","solutions":["Read the cause (e.getCause()) inside StreamException to identify the real AWS error and fix that root cause","Verify IAM permissions (kinesis:ListShards, kinesis:GetShardIterator, kinesis:GetRecords) and stream name/region config","Restart or retry the task if the cause is a transient AWS exception (throttling, timeout)","Upgrade AWS SDK retry settings / increase httpTimeout in the Kinesis consumer config"],"exampleFix":"// before\ncatch (Exception e) {\n  throw new StreamException(e);\n}\n// after\ncatch (Exception e) {\n  if (AWSClientUtil.isClientExceptionRecoverable((SdkException) e.getCause())) {\n    // retry with backoff instead of failing the task\n  }\n  throw new StreamException(e);\n}","handlingStrategy":"try-catch","validationCode":"// preflight: verify stream and permissions before starting\nGetStreamSummaryResponse s = kinesisClient.describeStream(DescribeStreamRequest.builder().streamName(stream).build());\nif (!s.streamDescriptionSummary().streamStatus().equals(\"ACTIVE\")) throw new IllegalStateException(\"stream not ACTIVE\");","typeGuard":"static boolean isRecoverableCause(StreamException e) {\n  return e.getCause() instanceof SdkException\n      && AWSClientUtil.isClientExceptionRecoverable((SdkException) e.getCause());\n}","tryCatchPattern":"try {\n  supplier.getPartitionIds(stream);\n} catch (StreamException e) {\n  if (isRecoverableCause(e)) { /* retry with backoff */ }\n  else { LOG.error(e.getCause(), \"unrecoverable Kinesis error\"); throw e; }\n}","preventionTips":["Always inspect StreamException.getCause() to find the real AWS error","Preflight IAM permissions with a cheap describeStream/ListShards call at task start","Set generous AWS SDK timeouts and retry policy in the kinesis consumer config","Alert on throttling metrics (ThrottlingException causes) rather than failing tasks"],"tags":["kinesis","aws","exception-wrapping"],"backgroundTag":"upstream-api-error","analyzedSha":"9b90983fd291f26935af934383ce360473179e4d","analyzedAt":"2026-09-07T13:32:30.957Z","contentChangedAt":"2026-09-07T13:32:30.957Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}