使用KCL 2下特定碎片的记录。x（运动）

Question

我在Kinesis流中的一些特定分片下有一组记录。我正在使用KCL 2. x消费者从kinesis中消费记录，但问题是消费者正在从流中所有可用的分片中获取我的记录。所以有没有什么方法可以在配置configBuilder对象或KCL消费者时指定分片或它们的ID，以便只消费来自指定分片的记录。

示例代码：

configsBuilder = new ConfigsBuilder(
        applicationName,
        streamName,
        kinesisAsyncClient,
        dynamoDbClient,
        cloudWatchClient,
        workerID,
        new RecordProcessorFactory());

scheduler = new Scheduler(
        configsBuilder.checkpointConfig(),
        configsBuilder.coordinatorConfig(),
        configBuilder.leaseManagementConfig(),
        configsBuilder.lifecycleConfig(),
        configsBuilder.metricsConfig(),
        configsBuilder.processorConfig(),
        configBuilder.retrievalConfig()
    );

    // start the kinesis records consumer.
    schedulerThread = new Thread(scheduler);
    schedulerThread.setDaemon(true);
    schedulerThread.start();

提前感谢！

Answer 1

KCL 2。x提供了一个ShardPrioritization界面，允许对碎片进行优先级排序或过滤：

/**
 * Provides logic to prioritize or filter shards before their execution.
 */
public interface ShardPrioritization {

    /**
     * Returns new list of shards ordered based on their priority.
     * Resulted list may have fewer shards compared to original list
     * 
     * @param original
     *            list of shards needed to be prioritized
     * @return new list that contains only shards that should be processed
     */
    List<ShardInfo> prioritize(List<ShardInfo> original);
}

也就是说，您可以提供ShardPrioritify实现，它将只留下与您相关的分片。

之后，只需在协调器配置中指定您的优先级：

configsBuilder.coordinatorConfig
          .shardPrioritization(new CustomShardsPrioritixation())