apache/druid · error · StreamException

StreamException

Error message

StreamException

What it means

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.

Source

Thrown at extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java:123

  /**
   * We call getRecords with limit 1000 to make sure that we can find the first (earliest) record in the shard.
   * In the case where the shard is constantly removing records that are past their retention period, it is possible
   * that we never find the first record in the shard if we use a limit of 1.
   */
  private static final int GET_SEQUENCE_NUMBER_RECORD_COUNT = 1000;
  private static final int GET_SEQUENCE_NUMBER_RETRY_COUNT = 10;

  /**
   * Catch any exception and wrap it in a {@link StreamException}
   */
  private static <T> T wrapExceptions(Callable<T> callable)
  {
    try {
      return callable.call();
    }
    catch (Exception e) {
      throw new StreamException(e);
    }
  }

  private class PartitionResource
  {
    private final StreamPartition<String> streamPartition;

    // shardIterator points to the record that will be polled next by recordRunnable
    // can be null when shard is closed due to the user shard splitting or changing the number
    // of shards in the stream, in which case a 'EOS' marker is used by the KinesisRecordSupplier
    // to indicate that this shard has no more records to read
    @Nullable
    private volatile String shardIterator;
    private volatile long currentLagMillis;

    private final AtomicBoolean fetchStarted = new AtomicBoolean();
    private ScheduledFuture<?> currentFetch;

View on GitHub (pinned to 9b90983fd2)

Solutions

1. Read the cause (e.getCause()) inside StreamException to identify the real AWS error and fix that root cause
2. Verify IAM permissions (kinesis:ListShards, kinesis:GetShardIterator, kinesis:GetRecords) and stream name/region config
3. Restart or retry the task if the cause is a transient AWS exception (throttling, timeout)
4. Upgrade AWS SDK retry settings / increase httpTimeout in the Kinesis consumer config

Example fix

// before
catch (Exception e) {
  throw new StreamException(e);
}
// after
catch (Exception e) {
  if (AWSClientUtil.isClientExceptionRecoverable((SdkException) e.getCause())) {
    // retry with backoff instead of failing the task
  }
  throw new StreamException(e);
}

// preflight: verify stream and permissions before starting
GetStreamSummaryResponse s = kinesisClient.describeStream(DescribeStreamRequest.builder().streamName(stream).build());
if (!s.streamDescriptionSummary().streamStatus().equals("ACTIVE")) throw new IllegalStateException("stream not ACTIVE");

static boolean isRecoverableCause(StreamException e) {
  return e.getCause() instanceof SdkException
      && AWSClientUtil.isClientExceptionRecoverable((SdkException) e.getCause());
}

try {
  supplier.getPartitionIds(stream);
} catch (StreamException e) {
  if (isRecoverableCause(e)) { /* retry with backoff */ }
  else { LOG.error(e.getCause(), "unrecoverable Kinesis error"); throw e; }
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• Failed to initialize Glue catalog
• Couldn't generate AWS token.
• can't reschedule fetch records runnable, recordsResult is nu
• RuntimeException
• getPosition() is not supported in Kinesis

AI-assisted analysis of apache/druid@9b90983fd2 (2026-09-07). Data as JSON: /api/errors/9a0c60123011b806. Report an issue: GitHub.