presto-kinesis/src/main/java/io/prestosql/plugin/kinesis/KinesisRecordSet.java/KinesisRecordSet/KinesisRecordSet
class KinesisRecordSet:    KinesisRecordSet(
            KinesisSplit split,
            ConnectorSession session,
            KinesisClientProvider clientManager,
            List<KinesisColumnHandle> columnHandles,
            RowDecoder messageDecoder,
            KinesisConfig kinesisConfig)
    {
        this.split = requireNonNull(split, "split is null");
        this.session = requireNonNull(session, "session is null");
        requireNonNull(kinesisConfig, "kinesisConfig is null");
        long dynamoReadCapacity = kinesisConfig.getDynamoReadCapacity();
        long dynamoWriteCapacity = kinesisConfig.getDynamoWriteCapacity();
        long checkPointIntervalMillis = kinesisConfig.getCheckpointInterval().toMillis();
        this.isLogBatches = kinesisConfig.isLogBatches();

        this.clientManager = requireNonNull(clientManager, "clientManager is null");

        this.columnHandles = requireNonNull(columnHandles, "columnHandles is null");
        this.messageDecoder = messageDecoder;

        ImmutableList.Builder<Type> typeBuilder = ImmutableList.builder();

        for (KinesisColumnHandle handle : columnHandles) {
            typeBuilder.add(handle.getType());
        }

        this.columnTypes = typeBuilder.build();

        // Note: these default to what is in the configuration if not given in the session
        this.batchSize = getBatchSize(session);
        this.maxBatches = getMaxBatches(this.session);

        this.fetchAttempts = kinesisConfig.getFetchAttempts();
        this.sleepTime = kinesisConfig.getSleepTime().toMillis();

        this.checkpointEnabled = isCheckpointEnabled(session);
        this.lastReadSequenceNumber = null;

        // Initialize checkpoint related code
        if (checkpointEnabled) {
            AmazonDynamoDBClient dynamoDBClient = clientManager.getDynamoDbClient();
            String dynamoDBTable = split.getStreamName();
            int curIterationNumber = getIterationNumber(session);
            String sessionLogicalName = getCheckpointLogicalName(session);

            String logicalProcessName = null;
            if (sessionLogicalName != null) {
                logicalProcessName = sessionLogicalName;
            }

            shardCheckpointer = new KinesisShardCheckpointer(
                    dynamoDBClient,
                    dynamoDBTable,
                    split,
                    logicalProcessName,
                    curIterationNumber,
                    checkPointIntervalMillis,
                    dynamoReadCapacity,
                    dynamoWriteCapacity);

            lastReadSequenceNumber = shardCheckpointer.getLastReadSeqNumber();
        }
        else {
            shardCheckpointer = null;
        }
    }