从 KCL 2.x 迁移至 KCL 3.x

本主题提供了将消费端从 KCL 2.x 迁移至 KCL 3.x 的分步说明。KCL 3.x 支持 KCL 2.x 消费端进行就地迁移。在滚动迁移工作程序时，可继续使用 Kinesis 数据流中的数据。

步骤 1：先决条件

开始使用 KCL 3.x 之前，请确保已具备以下条件：

步骤 2：添加依赖项

如果您使用的是 Maven，请将以下依赖项添加到您的 pom.xml 文件中。确保将 3.x.x 替换为最新的 KCL 版本。

<dependency>
    <groupId>software.amazon.kinesis</groupId>
    <artifactId>amazon-kinesis-client</artifactId>
    <version>3.x.x</version> <!-- Use the latest version -->
</dependency>

如果使用 Gradle，请在 build.gradle 文件中添加以下信息。确保将 3.x.x 替换为最新的 KCL 版本。

implementation 'software.amazon.kinesis:amazon-kinesis-client:3.x.x'

可以在 Maven Central 存储库中查看最新版本的 KCL。

步骤 3：设置迁移相关配置

要从 KCL 2.x 迁移至 KCL 3.x，就必须设置以下配置参数：

configsBuilder.coordiantorConfig().clientVersionConfig(ClientVersionConfig.CLIENT_VERSION_CONFIG_COMPLATIBLE_WITH_2X)

下面是如何设置 CoordinatorConfig.clientVersionConfig 从 KCL 2.x 迁移至 3.x 的示例。您可以按照自己的具体需求，根据需要调整其他配置：

Scheduler scheduler = new Scheduler(
    configsBuilder.checkpointConfig(),
    configsBuilder.coordiantorConfig().clientVersionConfig(ClientVersionConfig.CLIENT_VERSION_CONFIG_COMPLATIBLE_WITH_2X),
    configsBuilder.leaseManagementConfig(),
    configsBuilder.lifecycleConfig(),
    configsBuilder.metricsConfig(),
    configsBuilder.processorConfig(),
    configsBuilder.retrievalConfig()
);

重要的是，由于 KCL 2.x 和 3.x 使用的负载均衡算法不同，消费端应用程序中的所有工作程序在给定时间均使用相同的负载均衡算法。如果使用不同的负载平衡算法运行工作程序，可能会导致负载分配不够理想，因为这两种算法是独立运行的。

借助于这种 KCL 2.x 兼容性设置，KCL 3.x 应用程序可以在兼容 KCL 2.x 的模式下运行，并使用 KCL 2.x 的负载平衡算法，直至消费端应用程序中的所有工作程序都升级到 KCL 3.x。迁移完成后，KCL 会自动切换到完整版 KCL 3.x 功能模式，并开始为所有正在运行的工作程序使用新的 KCL 3.x 负载平衡算法。

第 4 步：遵循 shutdownRequested() 方法实现的最佳实践

KCL 3.x 中推出了一项名为“优雅移交租约”的功能，在租约重新分配过程中将租约移交给其他工作程序时，该功能可最大程度减少对数据的再处理。其实现方法是在租约移交之前，对租约表中最后处理的序列号进行检查点操作。为确保优雅移交租约功能的正常运行，必须保证在 RecordProcessor 类的 shutdownRequested 方法中调用 checkpointer 对象。如果在 shutdownRequested 方法中未调用 checkpointer 对象，可以按照以下示例所示进行实现。

/**
 * Invoked when either Scheduler has been requested to gracefully shutdown
 * or lease ownership is being transferred gracefully so the current owner
 * gets one last chance to checkpoint.
 *
 * Checkpoints and logs the data a final time.
 *
 * @param shutdownRequestedInput Provides access to a checkpointer, allowing a record processor to checkpoint
 *                               before the shutdown is completed.
 */
public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) {
    try {
       // Ensure that all delivered records are processed 
       // and has been successfully flushed to the downstream before calling 
       // checkpoint
       // If you are performing any asynchronous processing or flushing to
       // downstream, you must wait for its completion before invoking
       // the below checkpoint method.
        log.info("Scheduler is shutting down, checkpointing.");
        shutdownRequestedInput.checkpointer().checkpoint();
    } catch (ShutdownException | InvalidStateException e) {
        log.error("Exception while checkpointing at requested shutdown. Giving up.", e);
    } 
}

步骤 5：检查 KCL 3.x 收集工作程序指标的先决条件

KCL 3.x 收集 CPU 利用率指标（例如工作程序的 CPU 利用率），以期均匀地平衡各工作程序之间的负载。消费端应用程序的工作程序可运行在 Amazon EC2、Amazon EKS、Amazon EKS 或 AWS Fargate 上。只有在满足以下先决条件时，KCL 3.x 才能从工作程序收集 CPU 利用率指标：

Amazon Elastic Compute Cloud(Amazon EC2)

Amazon EC2 上的 Amazon Elastic Container Service (Amazon ECS)

上的 Amazon ECSAWS Fargate

Amazon EC2 上的 Amazon Elastic Kubernetes Service (Amazon EKS)

上的 Amazon EKSAWS Fargate

第 6 步：更新 KCL 3.x 的 IAM 权限

必须向与 KCL 3.x 消费端应用程序有关的 IAM 角色或策略添加以下权限。这涉及对 KCL 应用程序使用的现有 IAM 策略进行更新。有关更多信息，请参阅 KCL 消费端应用程序所必需的 IAM 权限。

第 7 步：将 KCL 3.x 代码部署到工作程序

在设置了迁移所需的配置并完成之前所有的迁移清单之后，就可以构建代码并将其部署到工作程序中。

步骤 8：完成迁移

在部署 KCL 3.x 代码期间，KCL 将继续使用来自 KCL 2.x 的租约分配算法。成功将 KCL 3.x 代码部署到所有工作程序后，KCL 会自动检测部署情况，并根据工作程序的资源利用率切换到新的租约分配算法。有关新租约分配算法的更多详细信息，请参阅KCL 如何向工作程序分配租约并平衡负载。

在部署期间，可使用发送至 CloudWatch 的指标来监控迁移过程。您可以监控 Migration 操作下的指标。所有指标都是每个 KCL 应用程序的指标，并设置为 SUMMARY 指标级别。如果 CurrentState:3xWorker 指标的 Sum 统计数据与 KCL 应用程序的工作程序总数一致，则表示已成功完成向 KCL 3.x 的迁移。

KCL 在迁移期间提供回滚至 2.x 兼容模式的能力。成功迁移至 KCL 3.x 后，如果不再需要回滚，我们建议移除 CLIENT_VERSION_CONFIG_COMPATIBLE_WITH_2X 的 CoordinatorConfig.clientVersionConfig 设置。移除该配置之后，就不会从 KCL 应用程序发出与迁移相关的指标。