Kafka系列之：对源连接器的的Exactly-Once支持

一、背景

幂等生产者：

• 写入 Kafka 的应用程序重复记录的一个常见来源是自动生产者重试，其中生产者会在某些情况下将批次重新发送到 Kafka，即使该批次已由代理提交。KIP-98 允许用户将其生产者配置为幂等地执行这些重试，这样下游应用程序只能看到自动重新发送的生产者批次中任何消息的一个实例，即使该批次已由生产者多次发送到 Kafka。

交易生产者：

• 该 KIP 面向用户的变化的另一个主要组成部分是事务生产者的引入，它可以创建跨越多个主题分区的事务并对它们执行原子写入。此外，事务生产者能够“隔离”其他生产者，这涉及声明事务 ID 的所有权，并禁止具有相同事务 ID 的任何先前生产者实例能够对 Kafka 进行任何后续写入。这对于防止客户端应用程序的僵尸实例发送重复消息特别有用。

术语
此处使用“防护”作为通用术语来描述禁用某些类型的应用程序的旧实例或实例组。

• “隔离”事务生产者意味着禁止该生产者对 Kafka 执行任何进一步的写入操作（例如，实现此目的的一种方法是使用相同的事务 ID 实例化一个新的事务生产者）
• “隔离”一代源任务意味着禁止该代中的每个源任务向 Kafka 生成更多源记录，或提交更多源偏移量

“源分区”在这里有两个含义。它可以指：

• 源连接器从中读取并能够分配给单个任务的外部系统的一部分。例如，数据库中的表或 Kafka 源连接器（例如 MirrorMaker 2）中的主题分区
• 源任务向 Connect 框架提供的键/值对中的键以及源记录（例如，请参阅SourceRecord 构造函数），用于跟踪源分区的消费进度
• “源偏移量”可以用来指上述键/值对中以“源分区”为键的值，也可以指整个键/值对（与“offset”可以用来指主题分区和其中某个位置的组合，或者只是主题/分区/偏移三重奏中的最后一个坐标）

二、目标

接收器连接器已经可以实现一次性交付一段时间了，并且在具有独特键约束等功能的系统中，甚至相对容易。然而，鉴于 Kafka 缺乏允许轻松修复和/或防止重复的功能，对于源连接器来说情况并非如此，并且如果没有框架级别的调整，源连接器的一次性交付仍然是不可能的。

造成这种情况的主要原因有两个：

源偏移准确性：一旦源任务对应的源记录已成功发送到 Kafka， Connect 框架就会以可配置的时间间隔定期将源任务偏移写入内部 Kafka 主题。这为大多数源连接器提供了至少一次传送保证，但在源记录写入 Kafka 但工作线程在将这些记录的偏移量写入 Kafka 之前被中断的情况下，允许重复写入的可能性。

僵尸：某些场景可能会导致多个任务同时运行并为同一源分区（例如数据库表或 Kafka 主题分区）生成数据。这也可能导致框架产生重复的记录。

源连接器中重复记录的另一个常见来源是自动生产者重试。但是，用户已经可以通过将源连接器配置为通过工作人员级别 producer.enable.idempotence 或连接器级别 producer.override.enable.idempotence 属性使用幂等生产者来解决此问题。

为了支持源连接器的一次性交付保证，框架应该扩展为以原子方式将源记录及其源偏移写入 Kafka，并防止僵尸任务向 Kafka 生成数据。

通过这些更改，任何源连接器的每个源分区一次最多分配给一个任务，并且能够在启动时根据先前任务向框架提供的源偏移量恢复上游源的消耗，应该能够一次性交付。

目标

• 需要尽可能少地更改每个连接器。Connect 生态系统目前已经相当成熟，即使是像新 SourceTask 方法这样轻量级的东西也必须应用于潜在的数十个甚至数百个连接器。当前的提案不需要对现有连接器进行此类更改，只要它们正确使用源偏移量 API 即可。
• 最大限度地减少配置更改。此功能的实现可能会变得相当复杂，但用户应该很容易启用和理解。 例如， KIP-98添加了对 Kafka 及其 Java 客户端的一次性支持，并且仅向用户公开了 3 个生产者属性和 1 个消费者属性。当前的提案仅添加了五个面向用户的配置属性。
• 最大限度地减少用户的操作负担。此功能很可能会被大量使用，甚至可能在 Connect 的更高版本中默认启用。应尽可能少地支持各种 Connect 部署方式和环境，包括但不限于安全意识和资源意识环境。当前提案在安全意识环境中的要求定义明确且易于预期，并且不会显着改变 Connect 的资源分配情况。
• 最大限度地减少问题和潜在的问题。没有人喜欢精美的印刷品。如果我们能够解决 Connect 用户或 Connect 开发人员的常见用例，我们应该；为了更简单的设计或较小的更改而放弃某些内容只能作为最后的手段。即使有详细记录，用例中的已知差距（尤其是那些因连接器而异的差距）也可能不会传播给最终用户或被最终用户理解，并且可能被视为此功能质量的缺陷。
• 总的来说，让这个功能给人们带来使用的乐趣，而不是痛苦。在投票通过之前，这个问题还没有定论！

三、公共接口

四、连接器 API 扩展

此处任何新引入的接口、类等都将添加到工件中 connect-api ，因为它们将成为 Connect 公共 API 的一部分。

ExactlyOnceSupport 引入了一个新的 枚举：

ExactlyOnce

package org.apache.kafka.connect.source;
 
/**
 * An enum to represent the level of support for exactly-once delivery from a source connector.
 */
public enum ExactlyOnceSupport {
    /**
     * Signals that a connector supports exactly-once delivery.
     */
    SUPPORTED,
    /**
     * Signals that a connector does not support exactly-once delivery.
     */
    UNSUPPORTED
}

API SourceConnector 已扩展为允许开发人员指定其连接器是否支持一次性交付：
源连接器

package org.apache.kafka.connect.source;
 
public abstract class SourceConnector extends Connector {
    // Existing fields and methods omitted
 
    /**
     * Signals whether the connector supports exactly-once delivery guarantees with a proposed configuration.
     * Developers can assume that worker-level exactly-once support is enabled when this method is invoked.
     * The default implementation will return {@code null}.
     * @param connectorConfigs the configuration that will be used for the connector.
     * @return {@link ExactlyOnceSupport#SUPPORTED} if the connector can provide exactly-once support,
     * and {@link ExactlyOnceSupport#UNSUPPORTED} if it cannot. If {@code null}, it is assumed that the
     * connector cannot.
     */
    public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> connectorConfig) {
        return null;
    }
}

TransactionContext 引入了一个新 界面：

事务上下文

package org.apache.kafka.connect.source;
 
/**
 * Provided to source tasks to allow them to define their own producer transaction boundaries when
 * exactly-once support is enabled.
 */
public interface TransactionContext {
    /**
     * Request a transaction commit after the next batch of records from {@link SourceTask#poll()}
     * is processed.
     */
    void commitTransaction();
 
    /**
     * Request a transaction commit after a source record is processed. The source record will be the
     * last record in the committed transaction.
     * @param record the record to commit the transaction after.
     */
    void commitTransaction(SourceRecord record);
 
    /**
     * Requests a transaction abort the next batch of records from {@link SourceTask#poll()}. All of
     * the records in that transaction will be discarded and will not appear in a committed transaction..
     */
    void abortTransaction();
 
    /**
     * Requests a transaction abort after a source record is processed. The source record will be the
     * last record in the aborted transaction. All of the records in that transaction will be discarded
     * and will not appear in a committed transaction.
     * @param record the record to abort the transaction after.
     */
    void abortTransaction(SourceRecord record);
}

该 SourceTaskContext 接口经过扩展，为开发人员提供对 TransactionContext 实例的访问（Javadocs 大部分是从 上的现有文档复制的SinkTaskContext::errantRecordReporter ）：

源任务上下文

package org.apache.kafka.connect.source;
 
public interface SourceTaskContext {
    // Existing fields and methods omitted
 
    /**
     * Get a {@link TransactionContext} that can be used to define producer transaction boundaries
     * when exactly-once support is enabled for the connector.
     *
     * <p>This method was added in Apache Kafka 3.0. Source tasks that use this method but want to
     * maintain backward compatibility so they can also be deployed to older Connect runtimes
     * should guard the call to this method with a try-catch block, since calling this method will result in a
     * {@link NoSuchMethodException} or {@link NoClassDefFoundError} when the source connector is deployed to
     * Connect runtimes older than Kafka 3.0. For example:
     * <pre>
     *     TransactionContext transactionContext;
     *     try {
     *         transactionContext = context.transactionContext();
     *     } catch (NoSuchMethodError | NoClassDefFoundError e) {
     *         transactionContext = null;
     *     }
     * </pre>
     *
     * @return the transaction context, or null if the user does not want the connector to define
     * its own transaction boundaries
     * @since 3.0
     */
    default TransactionContext transactionContext() {
        return null
    }
}

ConnectorTransactionBoundaries 引入了一个新的 枚举：

源连接器

package org.apache.kafka.connect.source;
 
/**
 * An enum to represent the level of support for connector-defined transaction boundaries.
 */
public enum ConnectorTransactionBoundaries {
    /**
     * Signals that a connector can define its own transaction boundaries.
     */
    SUPPORTED,
    /**
     * Signals that a connector cannot define its own transaction boundaries.
     */
    UNSUPPORTED
}

API还 SourceConnector 通过第二种新方法进行了扩展，允许开发人员指定其连接器是否可以定义自己的事务边界：源连接器

package org.apache.kafka.connect.source;
 
public abstract class SourceConnector extends Connector {
    // Existing fields and methods omitted
 
    /**
     * Signals whether the connector can define its own transaction boundaries with the proposed
     * configuration. Developers must override this method if they wish to add connector-defined
     * transaction boundary support; if they do not, users will be unable to create instances of
     * this connector that use connector-defined transaction boundaries. The default implementation
     * will return {@code UNSUPPORTED}.
     * @param connectorConfigs the configuration that will be used for the connector
     * @return whether the connector can define its own transaction boundaries  with the given
     * config.
     */
    public ConnectorTransactionBoundaries canDefineTransactionBoundaries(Map<String, String> connectorConfig) {
        return ConnectorTransactionBoundaries.UNSUPPORTED;
    }
}

五、REST API验证

当用户今天提交新的连接器配置时，Connect 会采取措施确保连接器配置在将其写入配置主题之前有效。这些步骤是同步执行的，如果检测到任何错误或发生意外故障，则会在其请求的 HTTP 响应正文和状态中向用户报告。

此验证在以下端点上进行：

• POST /connectors/
• PUT /connectors/{connector}/config
• PUT /connector-plugins/{connectorType}/config/validate
• 将添加一些新的飞行前验证逻辑，以实现同样适用于这些端点（以及将来执行类似飞行前连接器配置验证的任何其他端点）的一次性逻辑。

当需要一次性支持时

• 如果将exact.once.support 连接器配置设置为 required ，则将查询连接器的 SourceConnector::exactlyOnceSupport 方法。
• 如果结果为 UNSUPPORTED ，则将报告错误，并包含 Excellent.once.support 属性，指出连接器不提供给定配置的 Exactly-Once 支持。
• 如果结果为 null ，则将报告一个错误，并包含 Excellent.once.support 属性，指出 Connect 无法确定连接器是否提供精确一次保证，并且用户应在继续操作之前仔细查阅连接器的文档，方法可能是设置“请求”该属性。
• 如果结果是 SUPPORTED ，则不会报错。

当需要连接器定义的事务边界时

• 如果 transaction.boundary 连接器配置设置为 Connector ，则将查询连接器的 SourceConnector::canDefineTransactionBoundaries 方法。
• 如果结果为 ConnectorTransactionBoundaries.UNSUPPORTED 或 null ，则 transaction.boundary 属性将报告错误，指出连接器不支持定义自己的事务边界，并且将鼓励用户使用不同的事务边界配置连接器类型。
• 如果结果是 ConnectorTransactionBoundaries.SUPPORTED ，则不会报告任何错误。

六、新指标

将添加三个新的任务级 JMX 属性：

MBean name: kafka.connect:type=source-task-metrics,connector=([-.\w]+),task=([\d]+)

七、计划变更

偏移读取

当为worker设置为启用exactly.once.source.support时，该worker用于从偏移量主题读取源偏移量的消费者将配置一个新的默认属性isolation.level=read_comfilled。这将导致它们忽略属于未提交事务的任何记录，但仍然消耗根本不属于事务的记录。

用户将无法在工作线程配置中使用属性consumer.isolation.level或在连接器配置中使用属性consumer.override.isolation.level显式覆盖此设置。如果他们尝试这样做，用户提供的值将被忽略，并且将记录一条警告消息，通知他们这一事实。

偏移（和记录）写入

当为worker设置为启用exactly.once.source.support时，该worker创建的所有源任务将使用事务生产者写入Kafka。他们提供给工作人员的所有源记录都将在事务中写入 Kafka。一旦提交该事务，当前批次的偏移量信息将被写入同一事务内的偏移量主题。然后，事务将被提交。这将确保当且仅当该批次的源记录也写入 Kafka 时，源偏移量才会提交到 Kafka。

用户将能够使用新引入的 transaction.boundary 连接器配置属性（在“公共接口”部分中更详细地描述）来配置连接器的事务边界。

一旦偏移量提交完成，如果连接器（隐式或显式）配置了单独的偏移量主题，则提交的偏移量也将使用非事务性生产者和工作人员的主体写入工作人员的全局偏移主题。这将在与任务工作和偏移提交线程分开的线程上处理，并且根本不应该阻塞或干扰任务。如果工作线程由于某种原因未能写入这些偏移量，它将无限期地重试，但不会使任务失败。这样做是为了促进“硬”降级以及用户从每个连接器偏移主题切换回全局偏移主题的情况。

任务生产者将获得一个事务 ID $groupId-${connector}-${taskId}，其中 $groupId是Connect集群的组ID，${connector} 是连接器的名称，并且${taskId} 是任务的 ID（从零开始）。用户将无法使用工作人员级别的 Producer.transactional.id 或连接器级别的 Producer.override.transactional.id 属性覆盖此属性。如果他们尝试这样做，用户提供的值将被忽略，并且将记录一条警告消息，通知他们这一事实。

对于恰好一次的源任务，工作线程级别的 offset.flush.timeout.ms 属性将被忽略。它们将被允许花费必要的时间来完成偏移量提交，因为此时失败的成本是使源任务失败。目前，所有源任务偏移量提交都发生在单个共享工作全局线程上。为了支持没有超时的源任务提交，同时也防止滞后任务破坏集群上其他任务的可用性，将修改工作线程以允许同时源任务偏移量提交。

任务将其所有记录刷新到 Kafka 所需的时间可能会长于事务超时。在这种情况下，用户可以采取一些补救措施来使连接器恢复健康：调整生产者配置以获得更高的吞吐量，增加连接器使用的生产者的事务超时，减少偏移提交间隔（如果使用间隔） -基于事务边界），或切换到 transaction.boundary 属性的轮询值。对于由于生产者事务超时而失败的任务，我们将在错误消息中包含这些步骤。

SourceTask记录提交API

SourceTask::commit 和 SourceTask::commitRecord 方法将在每次成功的偏移提交后立即调用，包括在任务关闭期间发生的生命周期结束偏移提交。首先，将为已提交批次中的每条记录调用 commitRecord，然后调用 commit。在成功提交偏移量和完成这些调用之间，工作人员或任务可能会突然死亡。

因此，不能保证每次成功提交偏移量后都会调用这些方法。然而，只要任务能够仅根据提供给 Connect 框架的偏移信息及其生成的每条记录来准确恢复，就不会损害一次性交付保证。这些方法可以而且应该仍然用于在保证记录交付后清理和放弃资源，但使用它们来跟踪偏移量将阻止连接器实现一次性支持。

每个连接器偏移主题
无论worker上的exactly.once.source.support属性的值如何，源连接器都将被允许使用自定义偏移主题，可通过offsets.storage.topic属性进行配置。

首先，随着 KIP-458 的出现，现在可以建立一个单独的 Connect 集群，其连接器每个都针对不同的 Kafka 集群（通过 Producer.override.bootstrap.servers、consumer.override.bootstrap.servers 和/或 admin.override.bootstrap.servers 连接器属性）。目前，源任务偏移量仍然在整个 Connect 集群的单个全局偏移量主题中进行跟踪；但是，如果任务的生产者用于写入源偏移量，并且该生产者的目标 Kafka 集群与 Connect 工作线程用于其内部主题的集群不同，则这种情况将不再可能。为了很好地支持这种情况，必须对每个连接器的偏移主题进行推理，以允许平滑升级和降级，而不会在偏移数据中产生不必要的大间隙（请参阅下面的“迁移”）。尽管在这种情况下，用户不一定需要控制每个连接器的偏移主题的名称（例如，工作人员可以简单地在连接器覆盖的 Kafka 集群上创建一个具有相同名称的偏移主题），但公开此级别一旦意识到每个连接器偏移主题的必要性，控制的增加就不会增加此设计的复杂性。这样做的好处是，它应该使用户可以更轻松地在安全环境中配置连接器，在安全环境中，连接器主体对其目标 Kafka 集群的访问权限可能受到限制。对于熟悉 Kafka Streams 的用户和开发人员来说，这是导致全局偏移主题不可能实现的主要差异化因素。

其次，它消除了潜在的安全漏洞，恶意连接器可能会破坏集群上其他连接器可用的偏移量信息。目前可以通过将连接器配置为与工作人员主体不同的主体，并且仅向工作人员主体授予对偏移量主题的写访问权限来防止这种情况。但是，由于现在将使用同一个生产者来写入偏移数据和写入源任务记录，因此该方法将不再起作用。相反，允许连接器使用自己的偏移量主题应该允许管理员维护其集群的安全性，尤其是在多租户环境中。

最后，它允许用户限制在偏移量主题上挂起交易所产生的影响。如果任务A和B使用相同的offsets主题，并且任务A在任务B启动之前在该offsets主题上发起事务，那么任务A在没有提交其事务的情况下突然死亡，任务B将不得不等待该事务超时在它可以读取到偏移量主题的末尾之前。如果任务 A 的事务超时设置得非常高（例如，为了适应高吞吐量的突发），这将阻止任务 B 长时间处理任何数据。尽管这种情况在某些情况下可能是不可避免的，但为每个连接器使用专用的偏移量主题应该允许集群管理员隔离偏移量主题上挂起事务的影响范围。这样，虽然同一个连接器的任务仍然可能互相干扰，但至少不会干扰其他连接器的任务。这对于大多数多租户环境来说应该足够了。

配置
用户只能配置每个连接器偏移主题的名称。其他属性（例如分区数量和复制因子）将从工作配置中派生。

添加了工作程序启动时创建这些主题（如果这些主题不存在）的功能，并使用以下属性来定义这些新主题的复制因子和分区数量：

config.storage.replication.factor – 默认为 3，值必须 >= 1
offset.storage.replication.factor – 默认为 3，值必须 >= 1
status.storage.replication.factor – 默认为 3，值必须 >= 1
offset.storage.partitions – 默认为 25，值必须 >= 1
status.storage.partitions – 默认为 5，值必须 >= 1

此外，所有工作线程配置将更改为与以下模式匹配的其他可选属性，并在尝试创建内部主题时将它们（不携带）传递给 Connect 所有工作线程创建的新主题请求：

托管Kafka集群
无论worker上的exactly.once.source.support属性的值是什么，如果连接器配置包含offsets.storage.topic属性的值，它将在它生成的Kafka集群上使用具有该名称的偏移量主题数据到（可能与托管工作人员的全局偏移量主题的数据不同）。

隐式使用
在某些情况下，可能会隐式配置每个连接器偏移主题。具体来说，如果连接器的配置不包含 offsets.storage.topic 属性，但确实包含覆盖的 bootstrap.servers 值，该值会导致连接器针对与其他 Kafka 集群不同的连接器，则会发生这种情况（启用一次源支持时）承载工作人员的全局偏移量主题的主题。

像这样的连接器将被称为“隐式配置”以使用单独的偏移量主题。

该主题的名称将与全局偏移主题的名称相同，该名称由工作配置中的 offsets.storage.topic 属性控制。

创建
如果连接器显式或隐式配置为使用单独的偏移量主题，但该主题尚不存在，则工作线程将在启动之前自动尝试为属于该连接器的任何任务或连接器实例创建主题。这将使用从连接器配置构建的管理客户端（使用各种 admin.override.* 连接器属性、admin.* 工作器属性和按优先级降序排列的顶级工作器属性）来完成，使用的逻辑与工作人员已将其用于其内部主题。

平滑迁移
当第一次创建单独的offsets主题时，它自然是空的。为了避免丢失可能存储在全局偏移量主题中的偏移量信息，当连接器或任务实例向工作器请求偏移量时，它将获得其单独的偏移量主题和工作器的偏移量信息的组合视图。全局偏移主题。将优先考虑单独偏移量主题中存在的偏移量信息。

例如，如果连接器的全局偏移主题中存储的偏移量为：

工作人员的全局偏移量主题中存在的偏移量

{
  "partition": {
    "subreddit": "apachekafka"
  },
  "offset": {
    "timestamp": "4761"
  }
}
{
  "partition": {
    "subreddit": "CatsStandingUp"
  },
  "offset": {
    "timestamp": "2112"
  }
}

连接器单独的偏移量主题中的偏移量是：
单独偏移量主题中存在的偏移量

{
  "partition": {
    "subreddit": "CatsStandingUp"
  },
  "offset": {
    "timestamp": "2169"
  }
}
{
  "partition": {
    "subreddit": "grilledcheese"
  },
  "offset": {
    "timestamp": "489"
  }
}

由工作线程传递到连接器的偏移量将是：
提交给任务的偏移量

{
  "partition": {
    "subreddit": "apachekafka"
  },
  "offset": {
    "timestamp": "4761" // No offset for this partition was present in the separate offsets topic, so the one in the global offsets topic is used instead
  }
}
{
  "partition": {
    "subreddit": "CatsStandingUp"
  },
  "offset": {
    "timestamp": "2169" // Preference is given to the offset for this partition that comes from the separate offsets topic
  }
}
{
  "partition": {
    "subreddit": "grilledcheese"
  },
  "offset": {
    "timestamp": "489"
  }
}

八、任务计数记录

配置主题中将使用一种新类型的记录，即“任务计数记录”。如果在使用新的任务配置集启动任何任务之前重新配置连接器，则此记录显式跟踪必须被隔离的任务生产者的数量，并且将隐式跟踪是否有必要在开始之前执行一轮隔离连接器的任务。

名为“reddit-source”的连接器（包含 11 个任务）的此记录示例如下所示：

键：“任务计数-reddit-源”
值：{“任务”：11}

如果连接器的最新任务计数记录位于配置主题中的最新任务配置集之后，则工作人员可以安全地为该连接器创建和运行任务。例如，如果配置主题的内容（由每条记录的键表示）如下：

偏移量 0：任务-reddit-source-0
偏移量 1：任务-reddit-source-1
偏移量2：commit-reddit-source
偏移量 3：task-count-reddit-source

然后工作人员将能够安全地为 reddit 源连接器启动任务。

但是，如果配置主题的内容是这样的：

偏移量 0：任务计数-reddit-源
偏移量 1：任务-reddit-source-0
偏移量 2：任务-reddit-source-1
偏移量3：commit-reddit-source

新的内部端点将添加到 REST API 中：

PUT /连接器/{连接器}/栅栏

该端点将通过 KIP-507：保护内部连接 ​​REST 端点中引入的会话密钥机制进行保护，并且仅用于工作人员间通信；用户不应该直接查询它。无论worker的exactly.once.source.support值是多少，它都可用。

当工作人员收到对此端点的请求时，它将：

• 检查一下是否是领导者。如果不是领导者，则要么将请求转发给领导者，要么使请求失败，最多使用两跳。这种两跳策略已经在 Connect 中为其他需要转发请求的 REST 端点实现了，因此这里不再详细描述。
• 验证请求 URL 中的连接器是否存在并且是源连接器；如果不是，则请求失败。
• 阅读配置主题的末尾。
• 如果在连接器的最新任务配置集之后的配置主题中存在连接器的任务计数记录，则提供空体 200 响应。
• 如果配置主题中存在连接器的现有任务计数记录，并且该记录中的任务计数大于 1，或者连接器的新任务数大于 1*：
  • 通过使用连接器主体实例化管理客户端并调用 Admin::fenceProducers（如下所述的新 API；请参阅用于隔离事务性生产者的管理 API），隔离对该连接器的先前任务实例可能仍处于活动状态的所有生产者。事务 ID 将是连接器的每个任务将使用的 ID，假设活动的任务数量与连接器的最新任务计数记录一样多。这将并行完成。
  • 如果领导者在此期间收到为连接器写入新任务配置的请求，则该轮防护将立即取消，并提供 HTTP 409 CONFLICT 响应。
• 将隔离连接器的新任务计数记录写入配置主题
• 读取配置主题的末尾以验证任务计数记录是否已成功写入。
• 提供空的 200 响应。

第 6 步将在单独的线程上进行，以避免阻塞工作线程的牧羊人蜱线程（这对于检测和处理重新平衡以及执行某些 REST 请求的工作至关重要）。完成后，步骤 7 到 9 将在工作人员的牧羊人蜱线程上处理，以确保工作人员能够准确地了解在步骤 6 期间是否已提交连接器的新任务配置。

• • 检查任务计数是为了避免对永久单任务连接器（例如 Debezium 的 CDC 源连接器）进行不必要的防护工作。如果连接器的最新任务计数记录显示一项任务，则只有一项任务需要隔离。而且，如果该连接器的新配置包含一个任务，则该新任务将自动隔离其单个前任任务，因为它们将使用相同的事务 ID。同样的逻辑不适用于多任务连接器，即使重新配置后任务数量不变；

九、重新平衡的准备

当将正好.once.source.support 设置为启用并且检测到连接器的一组新任务配置时，工作人员将抢先停止该连接器的源任务。更详细地说：

• 当触发重新平衡时，在重新加入集群组之前，工作人员将抢先停止所有源连接器的所有任务，这些连接器的任务配置出现在最新任务计数记录之后的配置主题中。此步骤对于保留一次性交付保证不是必需的，但应该为任务在被强制隔离之前提供一个合理的机会来优雅地关闭。
• 停止这些任务将并行完成（因为分布式工作线程已经完成了集体连接器/任务停止/启动），并且工作线程在正常任务关闭超时期限​​内无法停止的任何任务（默认为五秒，但可以通过task.shutdown.graceful.timeout.ms工作者属性控制）将被放弃并允许继续运行。如果/当领导者在一轮僵尸围栏期间将其生产者围出时（如下所述），它们将被禁用。
• 由于这项工作将并行完成，任务关闭的默认超时时间相当低（五秒），并且当前用于执行集体任务停止的线程数为 8，因此这不太可能妨碍工作人员的工作。能够在重新平衡超时（默认为 60 秒）超时之前重新加入组。

这部分是对牧民重新平衡机制的误解的结果，实际上是没有必要的。在重新平衡期间（重新）加入组之前，工作线程已对所有重新配置的任务（如果使用增量重新平衡）或所有任务（如果使用急切重新平衡）执行此抢先停止。

十、源任务启动

当为工作线程设置为启用正好.once.source.support 时，将采取额外的步骤来确保任务仅在安全时运行。

在工作线程启动源任务之前，它将首先向领导者的内部僵尸防护端点发送任务连接器的请求。如果该请求因任何原因失败，该任务将被标记为 FAILED 并且启动将中止。

一旦工作线程实例化了源任务的生产者，它将再次读取配置主题的末尾，并且如果已为该连接器生成一组新的任务配置，它将中止任务的启动。

总而言之，任务的启动过程将是：

1. Worker开始启动一个源任务（由于用户手动重启、重新平衡、重新配置等）
2. Worker向Leader发出连接请求；如果此请求因任何原因失败，则任务将被标记为 FAILED 并且启动将中止
3. Worker 读取配置主题的末尾，并验证在连接器的最新任务配置集之后现在是否存在连接器的任务计数记录；如果不是这种情况，则放弃任务启动*
4. 如果连接器隐式或显式配置为使用单独的偏移量主题：
   Worker尝试创建主题（如果主题已经存在，则默默地吞下错误）；如果由于不可接受的原因而失败（使用与工作线程在创建其内部主题时相同的逻辑来区分可接受和不可接受的错误），则任务将被标记为 FAILED 并且启动将中止
5. Worker 为任务实例化事务生产者
6. Worker 读取配置主题的末尾
7. 如果此后为连接器生成了一组新的任务配置，则放弃任务启动*
   否则，开始轮询任务以获取数据

• • 如果发生这种情况，将自动启动一个新任务来代替该任务，以响应配置主题中的新任务配置集。用户无需执行任何操作（例如重新启动任务）。

十一、领导者访问配置主题

当将正好.once.source.support 设置为为某个工作程序准备或启用时，如果该工作程序是集群的领导者，它现在将使用事务生产者来保证最多一个工作程序能够写入配置主题随时。更详细地说：

• 重新平衡后，如果一个worker发现自己成为集群的leader，它会实例化一个事务生产者，其事务ID为 connect-cluster-${groupId} ，其中 ${groupId } 是集群的组ID。如果用户尝试通过在工作程序配置中设置 transactional.id 属性来覆盖此设置，则用户提供的值将被忽略，并且将记录一条警告消息，通知他们这一事实。工作线程将使用这个生产者来执行它在配置主题上执行的所有写入操作。它将开始并为其写入的每条记录提交一个事务。这可能看起来不寻常——如果事务并不是真正必要的话，为什么要使用事务性生产者？——但它确保只有最新的领导者才能够生产配置主题，而僵尸领导者（尽管可能很少见）会不能够。
• 例如，如果领导者在尝试为连接器排除上一代任务生产者时停滞不前，那么它可能会脱离组并成为僵尸。如果此时另一个工作进程被选举为领导者，然后重新配置连接器，则僵尸领导者可能会解除阻塞，然后在一组新的任务配置写入主题后尝试将任务计数记录写入配置主题。这会破坏配置主题的状态并违反一个关键假设：连接器的任务配置之后出现的任务计数记录意味着在启动该连接器的任务之前没有必要隔离上一代生产者。
• 通过在领导者上使用事务生产者，我们可以保证，如果从那时起没有其他工作人员成为领导者并为该主题的连接器写入新的任务配置，那么领导者只能将任务计数记录写入配置主题。我们还解决了 Connect 中现有的边缘情况，其中僵尸领导者可能会向配置主题写入不正确的信息。

十二、用于隔离事务生产者的管理 API

Java API 添加
管理界面将进行扩展，以包含通过事务 ID 隔离生产者的新方法。通过使用一个或多个事务性生产者并调用其 initTransactions 方法，已经可以实现相同的功能，但在管理客户端上进行了复制，以便更容易用于其他情况（有关更多详细信息，请参阅仅 Fencing 生产者拒绝的替代方案）：

Admin.java

public interface Admin {
 
    /**
     * Fence out all active producers that use any of the provided transactional IDs.
     *
     * @param transactionalIds The IDs of the producers to fence.
     * @param options          The options to use when fencing the producers.
     * @return The FenceProducersResult.
     */
    FenceProducersResult fenceProducers(Collection<String> transactionalIds,
                                        FenceProducersOptions options);
 
    /**
     * Fence out all active producers that use any of the provided transactional IDs, with the default options.
     * <p>
     * This is a convenience method for {@link #fenceProducers(Collection, FenceProducersOptions)}
     * with default options. See the overload for more details.
     *
     * @param transactionalIds The IDs of the producers to fence.
     * @return The FenceProducersResult.
     */
    default FenceProducersResult fenceProducers(Collection<String> transactionalIds) {
        return fenceProducers(transactionalIds, new FenceProducersOptions());
    }
 
}

API 的新类也将被添加。

FenceProducersResult 类：

FenceProducersResult.java

package org.apache.kafka.clients.admin;
 
/**
 * The result of the {@link Admin#fenceProducers(Collection)} call.
 *
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class FenceProducersResult {
 
    /**
     * Return a map from transactional ID to futures which can be used to check the status of
     * individual fencings.
     */
    public Map<String, KafkaFuture<Void>> fencedProducers() {
        // Implementation goes here
    }
 
    /**
     * Returns a future that provides the producer ID generated while initializing the given transaction when the request completes.
     */
    public KafkaFuture<Long> producerId(String transactionalId) {
        // Implementation goes here
    }
 
    /**
     * Returns a future that provides the epoch ID generated while initializing the given transaction when the request completes.
     */
    public KafkaFuture<Short> epochId(String transactionalId) {
        // Implementation goes here
    }
 
    /**
     * Return a future which succeeds only if all the producer fencings succeed.
     */
    public KafkaFuture<Void> all() {
        // Implementation goes here
    }
 
}

以及 FenceProducersOptions 类：
FenceProducersOptions.java

package org.apache.kafka.clients.admin;
 
/**
 * Options for {@link Admin#fenceProducers(Collection, FenceProducersOptions)}
 *
 * The API of this class is evolving. See {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class FenceProducersOptions extends AbstractOptions<FenceProducersOptions> {
    // No options (yet) besides those inherited from AbstractOptions
}

执行
在底层，这将通过 FindCoordinator 协议查找每个事务 ID 的事务协调器，然后通过调用没有生产者 ID 和生产者纪元的 InitProducerId 协议来使用该 ID 初始化新事务来实现。

• 提高 PID 的纪元，以便生产者的任何先前的僵尸实例都被隔离并且无法继续其事务。
• 恢复（前滚或回滚）生产者的前一个实例留下的任何未完成的事务。

ACL
此新 API 所需的 ACL 将与为每个指定事务 ID 使用事务生产者所需的 ACL 相同。具体来说，这相当于对 TransactionalId 资源上的 Write 和Describe 操作的授权，以及对 Cluster 资源上的 IdempoteWrite 操作的授权。

请注意，IdempotWrite ACL 从 2.8 开始已被弃用（请参阅 KIP-679），并且仅对于在 2.8 之前的 Kafka 集群上运行的 Connect 集群是必需的。

限制

• 需要分布式模式；一次性源连接器尚不支持独立模式
• 只能对所有源连接器启用或不启用 Exactly-once 源支持；它不能在每个连接器的基础上进行切换
• 为了实现一次性交付，连接器必须一次将源分区最多分配给一个任务（否则可能会发生重复写入）
• 为了实现一次性交付，连接器必须使用 Connect 源偏移 API 来跟踪进度（否则可能会发生重复写入或丢失记录）

十三、解决了故障/退化场景

这些场景不会损害集群的一次性交付保证。

在僵尸防护期间，Leader 无法将任务计数记录写入配置主题

在一轮僵尸防护期间，领导者将无限期地尝试将任务计数写入配置主题，并在写入后通过从主题读回来验证它是否已写入。

如果花费的时间太长，工人可能会从小组中退出，并且可能会选举出新的领导者。如果（现已被废黜的）领导者在此过程中变得畅通无阻并尝试写入配置主题，它将被其替代者隔离。如果它在读取配置主题时被阻止，它将醒来并完成响应服务。在牧羊人蜱线程的下一次迭代中，它将发现它已脱离组并尝试重新加入。

如果读取/写入配置主题的尝试没有被阻止，而是由于某种原因失败，领导者将立即离开组，然后尝试重新加入，提供 HTTP 500 ERROR 响应，其中包含交互失败的堆栈跟踪与配置主题。

在僵尸围栏期间无法将生产者拒之门外
例如，如果领导者无法联系或找到事务协调器，或者连接器的管理主体缺乏隔离生产者执行其任务所需的 ACL，则领导者可能无法在一轮僵尸隔离期间隔离生产者。

如果发生这种情况，领导者将提供 HTTP 500 ERROR 响应，其中包含未能隔离生产者的堆栈跟踪。

对于某些错误（例如由 ACL 不足引起的错误），堆栈跟踪将包含一条有用的错误消息，其中包含有关如何修复故障的说明（例如授予连接器的管理员主体权限，以使用适用的列表或描述来隔离生产者）交易 ID）。

收到此 500 响应的工作线程将在状态主题中标记任务对象 FAILED，并使用响应中包含的堆栈跟踪来填充状态消息的跟踪字段。

手动重新启动隔离任务

如果任务由于其生产者已被隔离而失败，用户可以尝试通过 REST API 手动重新启动该任务。如果工作人员对配置主题的视图是最新的，这只会导致重新启动的任务替换同一任务的任何现有实例，并且将保留一次性交付保证。但是，如果工作线程的配置主题视图已过时，则如果该任务最终从已分配给最新任务配置集中的不同任务的源分区生成数据，则重新启动该任务可能会损害一次性保证对于连接器。

为了防止手动用户重新启动损害一次性交付保证，Connect 工作人员将在继续任何源任务重新启动之前阅读配置主题的末尾。如果工作线程无法读取到配置主题的末尾，则重新启动请求将收到 HTTP 500 TIMEOUT 响应。如果工作线程在读取主题末尾时发现任务不再存在（因为连接器已被删除或其任务计数减少），它将提供通常的 HTTP 404 NOT FOUND 响应。如果由于从配置主题读取新信息而需要重新平衡，则将提供 HTTP 409 CONFLICT 响应。

Worker 在被分配源任务之后但在实例化生产者之前出现滞后
如果为工作人员分配了一个源任务，并在该连接器的最新一组任务配置之后在配置主题中看到任务计数记录，则假设该连接器的所有先前任务生成的所有生产者都已被隔离。而且，在领导者进行击剑时，这个假设可能成立。但是，工作人员可能会被分配一个源任务，观察配置主题中的任务计数记录，表明可以安全地启动该任务，然后在能够构造生产者之前阻塞一段较长的时间为了那个任务。例如，考虑一个场景，其中工作人员 F（追随者）、O（其他追随者）和 L（领导者）在源连接器 C 的任务 T 上进行操作：

1. 连接器 C 已重新配置，并且新的任务配置已写入配置主题
2. 触发重新平衡，worker L 将任务 T 分配给worker F
3. 工作人员 F 收到分配的任务，并在向工作人员 L 请求成功一轮防护后，能够从配置主题读取连接器 C 的最新任务计数记录
4. 工作线程 F 在能够实例化任务 T 的事务生产者之前会阻塞
5. 连接器 C 已重新配置，并且新的任务配置已写入配置主题
6. 触发重新平衡，由于 F 已脱离组，工作人员 L 将任务 T 分配给工作人员 O
7. 工作人员 O 收到其分配，并在向工作人员 L 请求成功一轮防护后，能够从配置主题读取连接器 C 的最新任务计数记录，实例化任务 T 的事务生产者，然后开始处理数据从那个任务
8. 工作线程 F 解除阻塞，实例化任务 T 的事务生产者，然后开始处理该任务的数据

在这种情况下，如果最初分配给任务 T 的源分区在步骤 5 的重新配置期间被重新分配给不同的任务，则工作线程 F 在步骤 8 中创建的任务实例将开始产生重复数据，并且一次性交付保证将受到损害。

这是可能的，因为在领导者执行隔离时，不能保证所有可能对连接器处于活动状态的生产者都已创建。如果生产者尚未创建，则无法将其排除在外；如果它是在隔离发生之后创建的，则必须采取不同的方法来确保不会产生重复数据。

在为源任务启动事务生产者之后，工作线程应通过对配置主题末尾的附加读取来覆盖这种情况。如果在工作线程被阻止并且发生一轮生产者防护后重新配置连接器，则工作线程将为其源任务启动事务性生产者，但随后发现连接器重新配置并中止启动。如果工作线程启动了事务性生产者，然后在完成对配置主题末尾的读取之前被阻塞，则领导者发起的这一轮生产者围栏应该隔离该生产者，并且任何后续的重新启动尝试都将阻塞，直到/除非工作线程能够完成对配置主题末尾的读取（并处理在此过程中从主题读取的新信息所需的任何重新平衡）。

意外的任务提交丢失的消息
如果事务生产者成功实例化，它不会再次尝试联系代理，直到尝试实际发送记录。这可能会导致有趣的行为：生产者启动并能够与事务协调器进行初始联系，然后整个 Kafka 集群死亡，然后生产者开始并提交事务，而不尝试发送任何记录，并且没有异常或其他情况生产者在此事件序列中的任何点生成失败指示。

什么时候可能会发生这种情况？如果任务配置了激进的 SMT，会删除给定批次中的所有记录，则其生产者在提交当前事务之前绝不会尝试发送任何记录。并且，一旦发生这种情况，工作线程将调用 SourceTask::commit，这可能会导致任务从上游系统删除数据（例如，通过确认来自 JMS 代理的一批消息）。即使发生这种情况，也不应该成为问题：可以从上游系统中删除本应由连接器删除的消息，无论这些记录的源偏移量是否已提交给 Kafka，因为最终结果是无论哪种方式都一样。

十四、允许的故障场景

这些场景可能会损害集群的一次性交付保证。

异构集群

如果工作线程处于活动状态并且不支持精确一次传送（因为未将正好.once.source.support设置为启用，或者工作线程正在运行旧版本的 Connect，该功能不可用），整个集群提供一次性保证的能力将受到损害。没有办法排除不合规的工人。即使开发出来了，问题也只会略有改变：如果一个工作人员处于活动状态并且无法被集群中的其他工作人员隔离，那么我们将处于与以前完全相同的位置。

这种失败场景需要在公共文档中指出，以便用户了解他们有责任确保没有不合规的工作人员处于活动状态。

非工人、非领导者生产者配置主题

如果除有效的领导者工作人员之外的任何内容最终写入配置主题，则集群依赖于确定启动任务是否安全以及是否需要隔离的所有假设都可能无效。在相当安全的 Connect 集群中，这种情况永远不应该发生，如果发生了，与破坏配置主题的其他后果相比，受损的交付保证可能会显得苍白无力。

十五、兼容性、弃用和迁移计划

与再平衡协议的互操作性

通过急切和增量合作再平衡协议都可以实现 Exactly-once 支持。支持这两种协议的额外设计复杂性是最小的，并且考虑到 Connect 框架仍然支持这两种协议，一些用户可能仍然会使用急切的重新平衡。如果支持这两种协议的工作不是令人望而却步，他们就不必升级到较新的重新平衡协议来获得此功能。

Worker主体权限
在将正好.once.source.support 设置为preparing 或enabled 之前，必须向worker 的生产者主体授予其写入的Kafka 集群的以下权限：

• • 请注意，IdempotWrite ACL 从 2.8 开始已被弃用，并且仅对于在 2.8 之前的 Kafka 集群上运行的 Connect 集群是必需的。

连接器主体权限
Producer
启用一次性源支持后，每个源连接器的生产者主体必须在其写入的 Kafka 集群上获得以下权限：

• • 请注意，IdempotWrite ACL 从 2.8 开始已被弃用，并且仅对于在 2.8 之前的 Kafka 集群上运行的 Connect 集群是必需的。

消费者
每个源连接器的消费者主体必须在其读取偏移量的 Kafka 集群上获得以下权限：

请注意，当前没有 Connect 用例需要为源连接器配置使用者主体。因此，此提议的更改在技术上为源连接器引入了“新”配置属性：以 Consumer.override 为前缀的消费者级别覆盖。

Admin
启用一次性源支持后，必须向连接器的管理主体授予其从中读取偏移量的 Kafka 集群的以下权限：

如果连接器的（隐式或显式配置的）偏移量主题尚不存在，则必须在将托管偏移量主题的 Kafka 集群上为其管理主体授予以下权限：

此外，无论如何，连接器的管理主体都必须在它读取偏移量的 Kafka 集群上获得以下权限：

为了确保 Connect 工作线程已读取到偏移量主题的末尾，这是必要的，即使在读取到末尾时该主题上存在打开的事务。这将通过以下方式完成：使用具有 read_uncommissed 隔离级别的管理客户端列出偏移量主题的结束偏移量，然后使用具有 read_commissed 隔离级别的该主题进行消费，直到至少管理客户端列出的那些偏移量为止。

十六、滚动升级

最多需要两次滚动升级才能在 Connect 集群上启用一次性源支持。

• 第一个滚动升级是将每个工作人员升级到可以提供一次性源支持的 Connect 版本，并将 just.once.source.support 设置为preparing。
• 第二个滚动升级将是在每个工作线程上实际启用一次源支持（通过将正好.once.source.support 设置为enabled）。

需要进行两次升级，以确保内部防护端点在每个工作人员需要之前都可用，并且在升级期间非领导工作人员无法写入配置主题。

十七、降级

可能有两种降级。可以通过将exact.once.source.support设置为disabled或为集群中的worker准备（“软”降级）来禁用对集群的exactly-once支持，或者可以将worker恢复为使用旧版本的Connect框架根本不支持一次性源（“硬”降级）。

软降级
软降级应该可以通过集群滚动来实现，其中每个工作线程都被关闭，其配置被更改为禁用一次性源支持，然后重新启动。在此过程中，所有连接器及其任务应继续运行和处理数据，这要归功于使滚动升级成为可能的相同逻辑。

硬降级
偏置精度
由于升级工作线程上的源任务偏移量仍会写入工作线程的全局偏移主题，因此即使降级工作线程不支持每个连接器偏移主题，它仍然可以获取其连接器的相对最新的源偏移量。其中一些偏移量可能已经过时或比连接器单独的偏移量主题中的偏移量更旧，但唯一的后果是连接器重复写入，这在没有启用一次性支持的任何集群上都是可能的。如果没有对全局偏移量主题进行任何写入，则自切换到专用偏移量主题以来连接器处理的所有记录都将在降级后重新处理，并可能导致大量重复项。虽然在技术上是允许的，因为在这种情况下，用户将故意切换到不支持一次性源连接器的 Connect 框架版本（因此容易受到记录的重复传递），但这种情况下的用户体验可能会受到影响。非常糟糕，因此 Connect 需要付出一些额外的努力来显着减少这种情况下降级的影响。

如果此时需要重新升级，则可能需要事先删除任何单独的每个连接器偏移主题。否则，工作人员将优先考虑现有的单独偏移量主题，即使该主题中的数据已过时并且全局偏移量主题中实际上存在更新的信息。

两步降级
执行硬降级的最安全方法是遵循与滚动升级相同的步骤，但相反。具体来说：

• 执行初始滚动降级，其中每个工作线程的正好.once.source.support 设置为 false。
• 执行第二次滚动降级，其中每个工作线程都被修改为使用早期版本的 Connect。

由于使滚动升级成为可能的相同逻辑，所有连接器和任务都应继续运行和处理数据。

单步降级
如果以单步滚动降级方式执行硬降级（即每个工作线程关闭后，立即降级到较低版本的 Connect），则某些任务可能会开始失败，因为其工作线程将无法访问已降级的工作人员的内部防护端点。降级完成后，重新启动这些任务就足以让它们再次运行。

十八、附加配置属性

这里引入的工作线程级别的正好.once.source.support和连接器级别的offsets.storage.topic、transaction.boundary、exact.once.support和transaction.boundary.interval.ms配置属性不应违反向后兼容性，因为它们都带有保留现有行为的默认值。从技术上讲，现有连接器可能会公开 offsets.storage.topic 配置属性，该属性现在将与新引入的框架属性发生冲突，但风险足够低，可以接受。

新引入的源连接器consumer.override。 -prefixed 属性在技术上也将向后不兼容，但出于同样的原因，这些属性最初是为接收器连接器和类似的 Producer.override 引入的。 - 和 admin.override。 -为源连接器引入了前缀属性，这应该是可以接受的。风险足够低。

十九、新的管理API

用于隔离生产者的新管理方法将使用现有协议（特别是 FindCoordinator 和 InitProducerId）来实现，因此不需要对 Kafka 二进制协议进行任何更改。因此，这些新方法应该适用于支持这些协议的任何代理。

二十、未来的工作

每个连接器的粒度
我们可能希望在每个连接器的基础上启用一次。这可以通过使用“合理的生产者”实例化每个源任务来实现，该生产者使用事务 ID 写入 Kafka，但实际上并不在事务内生成记录。这样，如果为该连接器启用了恰好一次，则其第一轮僵尸防护将能够隔离所有先前的生产者实例，即使它们不使用传统的事务性生产者。

独立模式支持
由于源记录及其偏移量的原子写入设计依赖于存储在 Kafka 主题中的源偏移量，因此独立模式不符合条件。如果有足够的需求，我们将来可能会将此功能添加到独立模式中。

默认启用
在该功能已可用于多个版本并且看起来足够稳定之后，我们可能希望默认启用该功能。

更保守的任务围栏
如果源分区在这些任务之间的划分不改变，则可能没有必要隔离上一代的所有任务。但是，不可能知道源分区到任务的实际分配，因为连接器没有 API 能够向 Connect 框架提供此信息。可以对 Connect 框架进行扩展以添加此 API，但留待将来工作。对于此类 API 需要考虑的一个重要问题是，它是否会考虑连接器本身不可见的配置更改，例如转换器或转换链更改。文章来源地址https://www.toymoban.com/news/detail-508619.html

到了这里，关于Kafka系列之：对源连接器的的Exactly-Once支持的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！