private SystemStreamMetadata createSystemStreamMetadata(String stream) {
    LOG.info("create stream metadata for stream {} based on aws stream", stream);
    Map<Partition, SystemStreamMetadata.SystemStreamPartitionMetadata> metadata = new HashMap<>();
    AmazonKinesisClient client = null;

    try {
      ClientConfiguration clientConfig = kConfig.getAWSClientConfig(system);
      AmazonKinesisClientBuilder builder = AmazonKinesisClientBuilder.standard()
          .withCredentials(kConfig.credentialsProviderForStream(system, stream))
          .withClientConfiguration(clientConfig);
      builder.setRegion(kConfig.getRegion(system, stream).getName());
      client = (AmazonKinesisClient) builder.build();
      StreamDescription desc = client.describeStream(stream).getStreamDescription();
      IntStream.range(0, desc.getShards().size())
          .forEach(i -> metadata.put(new Partition(i), SYSTEM_STREAM_PARTITION_METADATA));
    } catch (Exception e) {
      String errMsg = "couldn't load metadata for stream " + stream;
      LOG.error(errMsg, e);
      throw new SamzaException(errMsg, e);
    } finally {
      if (client != null) {
        client.shutdown();
      }
    }

    return new SystemStreamMetadata(stream, metadata);
  }
Source of truth for checkpointing is always kinesis and the offsets given by samza are always ignored by KCL.
Hence, return a placeholder for each ssp.
samza-aws/src/main/java/org/apache/samza/system/kinesis/KinesisSystemAdmin.java