kinesis-lib/src/main/java/com/streamsets/pipeline/stage/lib/kinesis/KinesisUtil.java
public static String getLastShardId(
      ClientConfiguration awsClientConfig,
      KinesisConfigBean conf,
      String streamName
  ) throws StageException {
    AmazonKinesis kinesisClient = getKinesisClient(awsClientConfig, conf);

    String lastShardId = null;
    try {
      StreamDescription description;
      do {
        if (lastShardId == null) {
          description = kinesisClient.describeStream(streamName).getStreamDescription();
        } else {
          description = kinesisClient.describeStream(streamName, lastShardId).getStreamDescription();
        }

        int pageSize = description.getShards().size();
        lastShardId = description.getShards().get(pageSize - 1).getShardId();

      } while (description.getHasMoreShards());

      return lastShardId;

    } finally {
      kinesisClient.shutdown();
    }
  }
public static String getLastShardId(
      ClientConfiguration awsClientConfig,
      KinesisConfigBean conf,
      String streamName
  ) throws StageException {
    AmazonKinesis kinesisClient = getKinesisClient(awsClientConfig, conf);

    String lastShardId = null;
    try {
      StreamDescription description;
      do {
        if (lastShardId == null) {
          description = kinesisClient.describeStream(streamName).getStreamDescription();
        } else {
          description = kinesisClient.describeStream(streamName, lastShardId).getStreamDescription();
        }

        int pageSize = description.getShards().size();
        lastShardId = description.getShards().get(pageSize - 1).getShardId();

      } while (description.getHasMoreShards());

      return lastShardId;

    } finally {
      kinesisClient.shutdown();
    }
  }
public static String getLastShardId ( ClientConfiguration awsClientConfig , KinesisConfigBean conf , String streamName ) throws StageException { AmazonKinesis kinesisClient = getKinesisClient ( awsClientConfig , conf ) ; String lastShardId = null ; try { StreamDescription description ; do { if ( lastShardId == null ) { description = kinesisClient . describeStream ( streamName ) . getStreamDescription ( ) ; } else { description = kinesisClient . describeStream ( streamName , lastShardId ) . getStreamDescription ( ) ; } int pageSize = description . getShards ( ) . size ( ) ; lastShardId = description . getShards ( ) . get ( pageSize - 1 ) . getShardId ( ) ; } while ( description . getHasMoreShards ( ) ) ; return lastShardId ; } finally { kinesisClient . shutdown ( ) ; } }
Get the last shard Id in the given stream
In preview mode, kinesis source uses the last Shard Id to get records from kinesis
@param awsClientConfig generic AWS client configuration
@param conf
@param streamName
Get the last shard Id in the given stream In preview mode kinesis source uses the last Shard Id to get records from kinesis