@Override
public ConnectorSplitSource getSplits(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorTableLayoutHandle layout, ConnectorSplitManager.SplitSchedulingStrategy splitSchedulingStrategy)
{
    KinesisTableLayoutHandle kinesislayout = handleResolver.convertLayout(layout);
    KinesisTableHandle kinesisTableHandle = kinesislayout.getTable();

    InternalStreamDescription desc = this.getStreamDescription(kinesisTableHandle.getStreamName());

    ImmutableList.Builder<ConnectorSplit> builder = ImmutableList.builder();
    for (Shard shard : desc.getShards()) {
        KinesisSplit split = new KinesisSplit(connectorId,
                kinesisTableHandle.getStreamName(),
                kinesisTableHandle.getMessageDataFormat(),
                shard.getShardId(),
                shard.getSequenceNumberRange().getStartingSequenceNumber(),
                shard.getSequenceNumberRange().getEndingSequenceNumber());
        builder.add(split);
    }

    return new FixedSplitSource(builder.build());
}
 

@Override
    public ConnectorSplitSource getSplits(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorTableLayoutHandle layout, SplitSchedulingContext splitSchedulingContext )
    {
        SplitSchedulingStrategy splitSchedulingStrategy = splitSchedulingContext.getSplitSchedulingStrategy();
        ElasticsearchTableLayoutHandle layoutHandle = (ElasticsearchTableLayoutHandle) layout;
        ElasticsearchTableHandle tableHandle = layoutHandle.getTable();

//        // Call out to our client to retrieve all tablet split metadata using the row ID domain and the secondary index
        List<ElasticsearchSplit> tabletSplits = client.getTabletSplits(session, tableHandle, layoutHandle, splitSchedulingStrategy); //tableHandle.getSerializerInstance()

        // Pack the tablet split metadata into a connector split
        return new FixedSplitSource(tabletSplits);
    }
 

@Override
public ConnectorSplitSource getSplits(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorTableLayoutHandle layout, SplitSchedulingStrategy splitSchedulingStrategy)
{
    HbaseTableLayoutHandle layoutHandle = (HbaseTableLayoutHandle) layout;
    HbaseTableHandle tableHandle = layoutHandle.getTable();

    String schemaName = tableHandle.getSchema();
    String tableName = tableHandle.getTable();
    String rowIdName = tableHandle.getRowId();

    // Get non-row ID column constraints
    List<HbaseColumnConstraint> constraints = getColumnConstraints(rowIdName, layoutHandle.getConstraint());

    // Get the row domain column range
    Optional<Domain> rDom = getRangeDomain(rowIdName, layoutHandle.getConstraint());

    // Call out to our client to retrieve all tablet split metadata using the row ID domain and the secondary index
    List<TabletSplitMetadata> tabletSplits = client.getTabletSplits(session, schemaName, tableName, rDom, constraints); //tableHandle.getSerializerInstance()

    // Pack the tablet split metadata into a connector split
    ImmutableList.Builder<ConnectorSplit> cSplits = ImmutableList.builder();
    for (TabletSplitMetadata splitMetadata : tabletSplits) {
        HbaseSplit split = new HbaseSplit(
                connectorId,
                schemaName,
                tableName,
                rowIdName,
                splitMetadata,
                constraints,
                tableHandle.getScanAuthorizations());
        cSplits.add(split);
    }

    return new FixedSplitSource(cSplits.build());
}
 

@Override
public ConnectorSplitSource getSplits(ConnectorTransactionHandle transactionHandle,
                                      ConnectorSession session, ConnectorTableLayoutHandle layout,
                                      SplitSchedulingStrategy splitSchedulingStrategy) {
    KuduTableLayoutHandle layoutHandle = (KuduTableLayoutHandle) layout;

    List<KuduSplit> splits = clientSession.buildKuduSplits(layoutHandle);

    return new FixedSplitSource(splits);
}
 

KinesisTableLayoutHandle convertLayout(ConnectorTableLayoutHandle layout)
{
    requireNonNull(layout, "layout is null");
    checkArgument(layout instanceof KinesisTableLayoutHandle, "layout is not an instance of KinesisTableLayoutHandle");
    KinesisTableLayoutHandle kinesisLayout = (KinesisTableLayoutHandle) layout;
    checkArgument(kinesisLayout.getConnectorId().equals(connectorId), "split is not for this connector");
    return kinesisLayout;
}
 

@Override
public ConnectorTableLayout getTableLayout(ConnectorSession session, ConnectorTableLayoutHandle handle)
{
    return new ConnectorTableLayout(handle);
}
 

@Override
public Class<? extends ConnectorTableLayoutHandle> getTableLayoutHandleClass()
{
    return ElasticsearchTableLayoutHandle.class;
}
 

@Override
public Class<? extends ConnectorTableLayoutHandle> getTableLayoutHandleClass()
{
    return HbaseTableLayoutHandle.class;
}
 

@Override
public ConnectorTableLayout getTableLayout(ConnectorSession session, ConnectorTableLayoutHandle handle)
{
    return new ConnectorTableLayout(handle);
}
 

@Override
public ConnectorTableLayout getTableLayout(ConnectorSession session, ConnectorTableLayoutHandle handle)
{
    ParaflowTableLayoutHandle layoutHandle = checkType(handle, ParaflowTableLayoutHandle.class, "tableLayoutHandle");
    return new ConnectorTableLayout(layoutHandle);
}
 

@Override
public Class<? extends ConnectorTableLayoutHandle> getTableLayoutHandleClass()
{
    return ParaflowTableLayoutHandle.class;
}
 

@Override
public ConnectorTableLayout getTableLayout(ConnectorSession session, ConnectorTableLayoutHandle handle) {
    return new ConnectorTableLayout(handle);
}
 

@Override
public Class<? extends ConnectorTableLayoutHandle> getTableLayoutHandleClass() {
    return EthereumTableLayoutHandle.class;
}
 

static EthereumTableLayoutHandle convertLayout(ConnectorTableLayoutHandle layout) {
    requireNonNull(layout, "layout is null");
    checkArgument(layout instanceof EthereumTableLayoutHandle, "layout is not an instance of EthereumTableLayoutHandle");
    return (EthereumTableLayoutHandle) layout;
}
 

@Override
public ConnectorTableLayout getTableLayout(ConnectorSession session, ConnectorTableLayoutHandle handle) {
    return new ConnectorTableLayout(handle);
}
 

@Override
public boolean supportsMetadataDelete(ConnectorSession session, ConnectorTableHandle tableHandle,
                                      ConnectorTableLayoutHandle tableLayoutHandle) {
    return false;
}
 

@Override
public Class<? extends ConnectorTableLayoutHandle> getTableLayoutHandleClass() {
    return KuduTableLayoutHandle.class;
}
 

@Override
public Class<? extends ConnectorTableLayoutHandle> getTableLayoutHandleClass()
{
    return KinesisTableLayoutHandle.class;
}
 

@Override
public ConnectorTableLayout getTableLayout(ConnectorSession connectorSession, ConnectorTableLayoutHandle connectorTableLayoutHandle)
{
    return new ConnectorTableLayout(connectorTableLayoutHandle);
}