配置 Streams 应用程序

（必填）应用程序 ID。每个流处理应用程序都必须具有唯一的 ID。必须将相同的 ID 提供给 应用程序的所有实例。建议仅使用字母数字字符 （dot）、（连字符） 和 （下划线）。例子：.-_"hello_world""hello_world-v1.0.0"

此 ID 用于以下位置，以将应用程序使用的资源与其他资源隔离开来：

• 作为默认的 Kafka 使用者和生成者前缀client.id
• 作为协调的 Kafka 使用者group.id
• 作为 state 目录中的子目录的名称 （cf.state.dir)
• 作为内部 Kafka 主题名称的前缀

提示：

当应用程序更新时，除非您想重用内部主题和状态存储中的现有数据，否则应该更改 。 例如，您可以将版本信息嵌入到 、 as 和 中。application.idapplication.idmy-app-v1.0.0my-app-v1.0.2

（必填）Kafka 引导服务器。这与底层生产者和使用者客户端用于连接到 Kafka 集群的设置相同。 例：。"kafka-broker1:9092,kafka-broker2:9092"

在将请求视为完成之前，领导必须收到的确认数。此控件 发送的记录的持久性。可能的值为：

• acks=0创建者不会等待服务器的确认，记录会立即添加到套接字缓冲区并被视为已发送。在这种情况下，不能保证 server 已经收到了记录，并且配置不会生效（因为 Client 端通常不会知道任何失败）。为每条记录返回的偏移量将始终设置为 .retries-1
• acks=1leader 将记录写入其本地日志并做出响应，而无需等待所有 follower 的完全确认。如果 leader 在确认记录后但在 followers 复制记录之前立即失败，则记录将丢失。
• acks=all领导节点等待完整的同步副本集确认记录。这可以保证如果至少有一个同步副本处于活动状态，则记录不会丢失。这是最强的可用保证。

有关更多信息，请参阅 Kafka 创建器文档。

要被视为已捕获并能够接收活动任务的实例的最大可接受滞后（要从更改日志中赶上的偏移总数）。Streams 将仅分配 状态存储位于可接受的恢复滞后范围内的实例（如果存在）的有状态活动任务，并分配预热副本以在后台恢复实例的状态 尚未赶上。对于给定工作负载，应对应于远低于一分钟的恢复时间。必须至少为 0。

注意：如果将此项设置为此项，则会有效地禁用预热副本和任务高可用性，从而允许 Streams 立即生成平衡的 分配任务并将其迁移到新实例，而无需先预热它们。Long.MAX_VALUE

默认的反序列化异常处理程序允许您管理无法反序列化的记录异常。这 可能是由数据损坏、序列化逻辑不正确或未处理的记录类型引起的。实现的异常 handler 需要返回 OR，具体取决于记录和引发的异常。返回将指示 Streams 应关闭，并指示 Streams 应忽略该问题 并继续处理。可以使用以下库内置异常处理程序：FAILCONTINUEFAILCONTINUE

• LogAndContinueExceptionHandler 的 此处理程序记录反序列化异常，然后向处理管道发出信号以继续处理更多记录。 这种记录和跳过策略允许 Kafka Streams 取得进展，而不是在有记录失败时失败 进行反序列化。
• LogAndFailExceptionHandler 的 Controller。 此处理程序记录反序列化异常，然后向处理管道发出信号以停止处理更多记录。

除了库提供的异常处理程序之外，您还可以提供自己的自定义异常处理程序来满足您的需求。例如，您可以选择转发 corrupt 记录放入隔离主题（想想：“死信队列”）中，以便进一步处理。为此，请使用 Producer API 将损坏的记录直接写入 隔离主题。更具体地说，您可以在 Streams 客户端之外创建一个单独的对象，并传入此对象 以及死信队列主题名称到 Map 中，然后可以从函数调用中检索。 这种方法的缺点是 “手动” 写入是 Kafka Streams 运行时库不可见的副作用。 因此，它们不会从 Streams API 的端到端处理保证中受益：KafkaProducerPropertiesconfigure

public class SendToDeadLetterQueueExceptionHandler implements DeserializationExceptionHandler {
    KafkaProducer<byte[], byte[]> dlqProducer;
    String dlqTopic;

    @Override
    public DeserializationHandlerResponse handle(final ErrorHandlerContext context,
                                                 final ConsumerRecord<byte[], byte[]> record,
                                                 final Exception exception) {

        log.warn("Exception caught during Deserialization, sending to the dead queue topic; " +
            "taskId: {}, topic: {}, partition: {}, offset: {}",
            context.taskId(), record.topic(), record.partition(), record.offset(),
            exception);

        dlqProducer.send(new ProducerRecord<>(dlqTopic, record.timestamp(), record.key(), record.value(), record.headers())).get();

        return DeserializationHandlerResponse.CONTINUE;
    }

    @Override
    public void configure(final Map<String, ?> configs) {
        dlqProducer = .. // get a producer from the configs map
        dlqTopic = .. // get the topic name from the configs map
    }
}

使用默认的 production 异常处理程序，可以管理在尝试与 broker 交互时触发的异常 例如尝试生成太大的记录。默认情况下，Kafka 提供并使用 DefaultProductionExceptionHandler，当这些异常发生时，它总是失败。

每个异常处理程序都可以返回 OR，具体取决于记录和引发的异常。返回将表示 Streams 应该关闭，并将表示 Streams 应忽略该问题并继续处理。如果要提供始终忽略太大记录的异常处理程序，则可以实现如下内容：FAILCONTINUEFAILCONTINUE

import java.util.Properties;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.common.errors.RecordTooLargeException;
import org.apache.kafka.streams.errors.ProductionExceptionHandler;
import org.apache.kafka.streams.errors.ProductionExceptionHandler.ProductionExceptionHandlerResponse;

public class IgnoreRecordTooLargeHandler implements ProductionExceptionHandler {
    public void configure(Map<String, Object> config) {}

    public ProductionExceptionHandlerResponse handle(final ErrorHandlerContext context,
                                                     final ProducerRecord<byte[], byte[]> record,
                                                     final Exception exception) {
        if (exception instanceof RecordTooLargeException) {
            return ProductionExceptionHandlerResponse.CONTINUE;
        } else {
            return ProductionExceptionHandlerResponse.FAIL;
        }
    }
}

Properties settings = new Properties();

// other various kafka streams settings, e.g. bootstrap servers, application id, etc

settings.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,
             IgnoreRecordTooLargeHandler.class);

时间戳提取器从 ConsumerRecord 的实例中提取时间戳。 时间戳用于控制流的进度。

默认提取程序为 FailOnInvalidTimestamp。 此提取器检索由 Kafka 生产者自动嵌入到 Kafka 消息中的内置时间戳 自 Kafka 版本 0.10 起的客户端。 根据 Kafka 服务器端代理和主题参数的设置， 此提取器为您提供：log.message.timestamp.typemessage.timestamp.type

• 事件时间处理语义 if 设置为 AK “producer time” （这是默认设置）。这表示 Kafka 生产者发送原始消息的时间。如果您使用 Kafka 的 官方生产者客户端，时间戳表示自 epoch 以来的毫秒数。log.message.timestamp.typeCreateTime
• ingestion-time processing semantics if 设置为 aka “broker 时间”。这表示 Kafka 代理收到原始消息的时间，以自纪元以来的毫秒为单位。log.message.timestamp.typeLogAppendTime

如果记录包含无效（即负数）内置 timestamp，因为 Kafka Streams 不会处理此记录，而是以静默方式删除它。无效的内置时间戳可能会 由于各种原因而发生：例如，如果您使用由 0.10 之前的 Kafka 生产者客户端写入的主题 或者由尚不支持新 Kafka 0.10 消息格式的第三方生产者客户端提供;另一种情况 这可能是在将 Kafka 集群从 升级到 之后发生的，其中生成的所有数据 with 不包含消息时间戳。FailOnInvalidTimestamp0.90.100.90.10

如果您的数据具有无效时间戳并希望对其进行处理，则有两个替代提取程序可用。 两者都适用于内置时间戳，但处理无效时间戳的方式不同。

• LogAndSkipOnInvalidTimestamp 的 此提取器记录一条警告消息，并将无效的时间戳返回给 Kafka Streams，Kafka Streams 不会处理，但会处理 静默删除记录。 这种对数和跳过策略允许 Kafka Streams 取得进展，而不是在存在 输入数据中的内置时间戳无效。
• UsePartitionTimeOnInvalidTimestamp 的 ValidTimestamp 中。 如果记录有效（即不负），则此提取器返回记录的内置时间戳。如果记录没有 具有有效的内置时间戳，则提取器会返回之前从 与当前记录相同的主题分区作为时间戳估计。如果无法估计时间戳，则 引发异常。

另一个内置提取器是 WallclockTimestampExtractor。 此提取器实际上并没有从消耗的记录中“提取”时间戳，而是返回 毫秒（想想：），这实际上意味着 Streams 将运行 根据事件的所谓处理时间。System.currentTimeMillis()

您还可以提供自己的时间戳提取器，例如检索嵌入在 payload 中的时间戳 消息。如果无法提取有效的时间戳，则可以引发异常、返回负时间戳或 估计时间戳。返回负时间戳将导致数据丢失 – 相应的记录不会 已处理，但无声地丢弃。如果要估计新的时间戳，可以使用 via 提供的值（即 Kafka Streams 时间戳估计）。以下是自定义实现的示例：previousTimestampTimestampExtractor

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.streams.processor.TimestampExtractor;

// Extracts the embedded timestamp of a record (giving you "event-time" semantics).
public class MyEventTimeExtractor implements TimestampExtractor {

  @Override
  public long extract(final ConsumerRecord<Object, Object> record, final long previousTimestamp) {
    // `Foo` is your own custom class, which we assume has a method that returns
    // the embedded timestamp (milliseconds since midnight, January 1, 1970 UTC).
    long timestamp = -1;
    final Foo myPojo = (Foo) record.value();
    if (myPojo != null) {
      timestamp = myPojo.getTimestampInMillis();
    }
    if (timestamp < 0) {
      // Invalid timestamp!  Attempt to estimate a new timestamp,
      // otherwise fall back to wall-clock time (processing-time).
      if (previousTimestamp >= 0) {
        return previousTimestamp;
      } else {
        return System.currentTimeMillis();
      }
    }
  }

}

然后，您将在 Streams 配置中定义自定义时间戳提取器，如下所示：

import java.util.Properties;
import org.apache.kafka.streams.StreamsConfig;

Properties streamsConfiguration = new Properties();
streamsConfiguration.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, MyEventTimeExtractor.class);

记录键的默认 Serializer/Deserializer 类，除非由用户设置，否则为 null。Kafka Streams 中的序列化和反序列化发生 每当需要具体化数据时，例如：

记录值的默认 Serializer/Deserializer 类，除非由用户设置，否则为 null。Kafka Streams 中的序列化和反序列化 每当需要具体化数据时都会发生，例如：

• 每当从 Kafka 主题读取数据或将数据写入 Kafka 主题时（例如，通过 and 方法）。StreamsBuilder#stream()KStream#to()
• 每当从状态存储读取数据或将数据写入状态存储时。

数据类型和序列化中对此进行了更详细的讨论。

窗口化键的内部类的默认 Serializer/Deserializer 类。Kafka Streams 中的序列化和反序列化发生 每当需要具体化数据时，例如：

窗口化值的内部类的默认 Serializer/Deserializer 类。Kafka Streams 中的序列化和反序列化发生 每当需要具体化数据时都会发生，例如：

• 每当从 Kafka 主题读取数据或将数据写入 Kafka 主题时（例如，通过 and 方法）。StreamsBuilder#stream()KStream#to()
• 每当从状态存储读取数据或将数据写入状态存储时。

数据类型和序列化中对此进行了更详细的讨论。

此配置设置从原始工作分配中移动任务的成本，该成本由 或 计算。䋰 它们控制优化器是倾向于最小化跨机架流量还是最小化现有分配中任务的移动。如果此配置设置为大于 的值，则 优化器将尝试维护 Task 分配器计算的现有分配。请注意，优化器会考虑这两个配置的比率，以支持维护现有分配或最小化流量成本。例如，设置为 10 和 1 比设置为 100 和 50 更有可能保持现有分配。StickyTaskAssignorHighAvailabilityTaskAssignorrack.aware.assignment.traffic_costrack.aware.assignment.traffic_costrack.aware.assignment.non_overlap_costrack.aware.assignment.traffic_costrack.aware.assignment.non_overlap_costrack.aware.assignment.traffic_cost

默认值为 null，这意味着将使用不同赋值器中的 default。在 中，它的默认值为 10，并且具有 默认值 1，这意味着在 中首选保持粘性。在 中，它的默认值为 1，并且具有 默认值 10，这意味着在 中首选最小化跨机架流量。non_overlap_costStickyTaskAssignorrack.aware.assignment.traffic_costStickyTaskAssignorHighAvailabilityTaskAssignorrack.aware.assignment.traffic_costHighAvailabilityTaskAssignor

此配置设置 Kafka Streams 用于机架感知任务分配的策略，以便可以减少从代理到客户端的交叉流量。此配置仅在 broker 上设置，并在 Kafka Streams 侧设置时生效。此配置有两个设置：broker.rackclient.rack

• none.这是默认值，这意味着将禁用机架感知任务分配。
• min_traffic.此设置意味着机架感知任务分配器将计算一个分配，以尝试最小化跨机架流量。
• balance_subtopology.此设置意味着机架感知任务分配器将计算一个分配，该分配将尝试将任务从同一子拓扑平衡到不同的客户端，并在此基础上最大限度地减少跨机架流量。

此配置可与 rack.aware.assignment.non_overlap_cost 和 rack.aware.assignment.traffic_cost 一起使用，以平衡减少跨机架流量和维持现有分配。

此配置设置用于在 Kafka Streams 之间分配备用副本的标签键列表 客户。配置后，Kafka Streams 将尽最大努力将备用任务分发到 具有不同 tag 值的 client。

可以通过 prefix 设置 Kafka Streams 客户端的标签。例：client.tag.

Client-1                                   | Client-2
_______________________________________________________________________
client.tag.zone: eu-central-1a             | client.tag.zone: eu-central-1b
client.tag.cluster: k8s-cluster1           | client.tag.cluster: k8s-cluster1
rack.aware.assignment.tags: zone,cluster   | rack.aware.assignment.tags: zone,cluster


Client-3                                   | Client-4
_______________________________________________________________________
client.tag.zone: eu-central-1a             | client.tag.zone: eu-central-1b
client.tag.cluster: k8s-cluster2           | client.tag.cluster: k8s-cluster2
rack.aware.assignment.tags: zone,cluster   | rack.aware.assignment.tags: zone,cluster

在上面的示例中，我们有四个 Kafka Streams 客户端，分别跨两个区域 （、） 和两个集群 （， ）。 对于位于 上的活动任务，Kafka Streams 将在 上分配备用任务，因为 具有不同于和不同的 。eu-central-1aeu-central-1bk8s-cluster1k8s-cluster2Client-1Client-4Client-4zoneclusterClient-1

此配置设置跨机架流量的成本。䋰 它们控制优化器是倾向于最小化跨机架流量还是最小化现有分配中任务的移动。如果此配置设置为大于 的值，则 优化器将尝试计算一个 assignment，以最小化跨机架流量。请注意，优化器会考虑这两个配置的比率，以支持维护现有分配或最小化流量成本。例如，设置为 10 和 1 比设置为 100 和 50 更有可能最大限度地减少跨机架流量。rack.aware.assignment.non_overlap_costrack.aware.assignment.non_overlap_costrack.aware.assignment.traffic_costrack.aware.assignment.non_overlap_costrack.aware.assignment.traffic_costrack.aware.assignment.non_overlap_cost

默认值为 null，这意味着将使用不同分配器中的默认流量成本。在 中，它的默认值为 1，默认值为 10。在 中，它的默认值为 10，默认值为 1。StickyTaskAssignorrack.aware.assignment.non_overlap_costHighAvailabilityTaskAssignorrack.aware.assignment.non_overlap_cost

此配置控制 Streams 将等待多长时间来获取数据，以便 提供 In-order 处理语义。

在处理具有多个输入分区的任务时（如在 join 或 merge 中）， Streams 需要选择要处理下一条记录的分区。 当所有输入分区都有本地缓冲数据时，Streams 会选择该分区 其下一条记录的时间戳最低。这具有理想的效果 按时间戳顺序整理 Importing 分区，这通常是 想要在流式处理中加入或合并。但是，当 Streams 没有任何数据时 在本地缓冲，则不知道下一个分区 record 的时间戳将低于或高于其余的 partitions 的记录。

有两种情况需要考虑：要么在 broker 尚未获取，或者 Streams 已完全赶上 分区，并且创建者根本没有生成任何新记录 自 Streams 轮询上一批以来。

默认值 会导致 Streams 在检测到任务没有本地 buffered data 的 Partition，但 broker 上有可用的数据。 具体来说，当本地缓冲区中存在空分区，但 Streams 该分区的滞后不为零。但是，一旦 Streams 赶上 代理，它将继续处理，即使其中一个 分区。也就是说，它不会等待生成新数据。 此默认值旨在牺牲一些吞吐量以换取直观 更正 join 语义。0

任何大于零的配置值都表示 Streams 在具有捕获但为空的分区时将等待的额外毫秒数。 换句话说，这是等待生成新数据的时间量 添加到输入分区中，以确保在事件中按顺序处理数据 一个缓慢的生产者。

的 config 值表示 Streams 在选择 按时间戳创建下一条记录，从而以 引入无序处理。-1

为了保持 任务在一个实例上可用，同时在另一个实例上预热，它已被重新分配到。用于限制额外代理的数量 traffic 和 cluster state 可用于实现高可用性。增加此值将允许 Streams 一次预热更多任务，从而加快时间 对于重新分配的预热，以恢复足够的状态，以便它们转换为活动任务。必须至少为 1。num.standbys

请注意，一个预热副本对应一个 Stream Task。此外，请注意，每个预热任务只能在 再平衡（通常在所谓的探测再平衡期间，以 Config 指定的频率发生）。这意味着 活动任务从一个 Kafka Streams 实例迁移到另一个实例的最大速率可以由以下因素决定 ( / ).probing.rebalance.interval.msmax.warmup.replicasprobing.rebalance.interval.ms

建议：

将备用数据库的数量增加到 1 个，以获得即时故障转移，即高可用性。 增加备用数据库的数量需要更多的客户端存储空间。 例如，如果 1 个备用数据库，则需要 2 倍空间。

注意：

如果开启 n 个备用任务，则需要预置 n+1 个实例。KafkaStreams

处理异常处理程序允许您管理在处理记录期间触发的异常。实现的异常 handler 需要返回 OR，具体取决于记录和引发的异常。返回将指示 Streams 应关闭，并指示 Streams 应忽略该问题 并继续处理。可以使用以下库内置异常处理程序：FAILCONTINUEFAILCONTINUE

• LogAndContinueProcessingExceptionHandler 中： 此处理程序记录处理异常，然后向处理管道发出信号以继续处理更多记录。 这种记录和跳过策略允许 Kafka Streams 在存在无法处理的记录时取得进展，而不是失败。
• LogAndFailProcessingExceptionHandler 的 ExceptionHandler 中。 此处理程序记录处理异常，然后向处理管道发出信号以停止处理更多记录。

除了库提供的异常处理程序之外，您还可以提供自己的自定义异常处理程序来满足您的需求。例如，您可以选择转发 corrupt 记录放入隔离主题（想想：“死信队列”）中，以便进一步处理。为此，请使用 Producer API 将损坏的记录直接写入 隔离主题。更具体地说，您可以在 Streams 客户端之外创建一个单独的对象，并传入此对象 以及死信队列主题名称到 Map 中，然后可以从函数调用中检索。 这种方法的缺点是 “手动” 写入是 Kafka Streams 运行时库不可见的副作用。 因此，它们不会从 Streams API 的端到端处理保证中受益：KafkaProducerPropertiesconfigure

public class SendToDeadLetterQueueExceptionHandler implements ProcessingExceptionHandler {
    KafkaProducer<byte[], byte[]> dlqProducer;
    String dlqTopic;

    @Override
    public ProcessingHandlerResponse handle(final ErrorHandlerContext context,
                                            final Record record,
                                            final Exception exception) {

        log.warn("Exception caught during message processing, sending to the dead queue topic; " +
            "processor node: {}, taskId: {}, source topic: {}, source partition: {}, source offset: {}",
            context.processorNodeId(), context.taskId(), context.topic(), context.partition(), context.offset(),
            exception);

        dlqProducer.send(new ProducerRecord<>(dlqTopic, null, record.timestamp(), (byte[]) record.key(), (byte[]) record.value(), record.headers()));

        return ProcessingHandlerResponse.CONTINUE;
    }

    @Override
    public void configure(final Map<String, ?> configs) {
        dlqProducer = .. // get a producer from the configs map
        dlqTopic = .. // get the topic name from the configs map
    }
}

建议：

虽然从技术上讲可以将 EOS 与任何复制因子一起使用，但使用复制因子的 比 3 有效地使 EOS 无效。因此，强烈建议使用 replication factor 3（与 一起）。此建议适用于所有主题（即 、、Kafka Streams 内部主题和用户主题）。min.in.sync.replicas=2__transaction_state__consumer_offsets

这指定了 Kafka Streams 在使用本地状态或流 重新分区以进行聚合。复制对于容错非常重要。没有复制，甚至单个代理失败 可能会阻止流处理应用程序的进度。建议使用与源主题类似的复制因子。

建议：

将复制因子增加到 3，以确保内部 Kafka Streams 主题最多可以容忍 2 个代理故障。 请注意，您还需要更多的存储空间（复制因子为 3 时为 3 倍）。

RocksDB 配置。Kafka Streams 使用 RocksDB 作为持久存储的默认存储引擎。更改默认值 配置，你可以通过 rocksdb.config.setter 实现和提供你的自定义类。RocksDBConfigSetter

下面是一个调整 RocksDB 消耗的内存大小的示例。

public static class CustomRocksDBConfig implements RocksDBConfigSetter {
    // This object should be a member variable so it can be closed in RocksDBConfigSetter#close.
    private org.rocksdb.Cache cache = new org.rocksdb.LRUCache(16 * 1024L * 1024L);

    @Override
    public void setConfig(final String storeName, final Options options, final Map<String, Object> configs) {
        // See #1 below.
        BlockBasedTableConfig tableConfig = (BlockBasedTableConfig) options.tableFormatConfig();
        tableConfig.setBlockCache(cache);
        // See #2 below.
        tableConfig.setBlockSize(16 * 1024L);
        // See #3 below.
        tableConfig.setCacheIndexAndFilterBlocks(true);
        options.setTableFormatConfig(tableConfig);
        // See #4 below.
        options.setMaxWriteBufferNumber(2);
    }

    @Override
    public void close(final String storeName, final Options options) {
        // See #5 below.
        cache.close();
    }
}

Properties streamsSettings = new Properties();
streamsConfig.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, CustomRocksDBConfig.class);

例如：

1. BlockBasedTableConfig tableConfig = (BlockBasedTableConfig) options.tableFormatConfig();获取对现有表配置的引用，而不是创建新的表配置，这样您就不会意外覆盖默认值，例如，这是一项重要的优化。BloomFilter
2. tableConfig.setBlockSize(16 * 1024L);根据 RocksDB GitHub 中的这些说明修改默认数据块大小。
3. tableConfig.setCacheIndexAndFilterBlocks(true);不要让 index 和 filter 块无限增长。有关更多信息，请参阅 RocksDB GitHub。
4. options.setMaxWriteBufferNumber(2);请参阅 RocksDB GitHub 中的高级选项。
5. cache.close();为避免内存泄漏，您必须关闭您构建的任何扩展 org.rocksdb.RocksObject 的对象。有关更多详细信息，请参阅 RocksJava 文档。