@Override
  public long getSplitBacklogBytes() {
    // Safety check in case a progress check is made for the start method is called.
    if (shardReadersPool == null) {
      return UnboundedReader.BACKLOG_UNKNOWN;
    }

    Instant latestRecordTimestamp = shardReadersPool.getLatestRecordTimestamp();

    if (latestRecordTimestamp.equals(BoundedWindow.TIMESTAMP_MIN_VALUE)) {
      LOG.debug("Split backlog bytes for stream {} unknown", spec.getStreamName());
      return UnboundedSource.UnboundedReader.BACKLOG_UNKNOWN;
    }

    if (latestRecordTimestamp.plus(spec.getUpToDateThreshold()).isAfterNow()) {
      LOG.debug(
          "Split backlog bytes for stream {} with latest record timestamp {}: 0 (latest record timestamp is up-to-date with threshold of {})",
          spec.getStreamName(),
          latestRecordTimestamp,
          spec.getUpToDateThreshold());
      return 0L;
    }

    if (backlogBytesLastCheckTime.plus(backlogBytesCheckThreshold).isAfterNow()) {
      LOG.debug(
          "Split backlog bytes for {} stream with latest record timestamp {}: {} (cached value)",
          spec.getStreamName(),
          latestRecordTimestamp,
          lastBacklogBytes);
      return lastBacklogBytes;
    }

    try {
      lastBacklogBytes = kinesis.getBacklogBytes(spec.getStreamName(), latestRecordTimestamp);
      backlogBytesLastCheckTime = Instant.now();
    } catch (TransientKinesisException e) {
      LOG.warn(
          "Transient exception occurred during backlog estimation for stream {}.",
          spec.getStreamName(),
          e);
    }
    LOG.info(
        "Split backlog bytes for {} stream with {} latest record timestamp: {}",
        spec.getStreamName(),
        latestRecordTimestamp,
        lastBacklogBytes);
    return lastBacklogBytes;
  }
@Test
  public void getSplitBacklogBytesShouldReturnUnknownIfNotStarted() {
    assertThat(reader.getSplitBacklogBytes()).isEqualTo(UnboundedReader.BACKLOG_UNKNOWN);
  }