在许多情况下，记录的时间戳（显式或隐式地）嵌入到记录本身。另外，用户可能希望例如基于包含当前事件时间水印的Kafka流中的特殊记录，周期性地或以不规则的方式发出水印。对于这些情况，Flink Kafka Consumer允许指定一个AssignerWithPeriodicWatermarks或一个AssignerWithPunctuatedWatermarks。

可以按此处所述指定自定义的时间戳提取器/水印发射器，或使用预定义的提取器/水印发射器 。这样做之后，可以通过以下方式将其传递给消费者：

在内部，每个Kafka分区都会执行分配器的实例。指定了这样的分配器后，对于从Kafka读取的每个记录，将调用extractTimestamp(T element, long previousElementTimestamp)，为记录分配时间戳， 并调用Watermark getCurrentWatermark()【周期性】或 Watermark checkAndGetNextWatermark(T lastElement, long extractedTimestamp)【标点】以确定是否应该发出新的水印，以及使用哪个水印时间戳记。

注意：如果水印分配器依靠从Kafka读取的记录来推进其水印（通常是这种情况），则所有主题和分区都需要具有连续的记录流。否则，整个应用程序的水印将无法使用，并且所有基于时间的操作（例如时间窗口或带有计时器的功能）都无法取得进展。单个空闲的Kafka分区会导致此行为。计划对Flink进行改进以防止这种情况的发生（请参阅FLINK-5479：FlinkKafkaConsumer中的按分区水印应考虑空闲分区）。同时，一种可能的解决方法是将心跳消息发送到所有消耗的分区，以提高空闲分区的水印。

Flink的Kafka Producer被调用FlinkKafkaProducer011（或010用于Kafka 0.10.0.x版本等，或仅FlinkKafkaProducer用于Kafka> = 1.0.0版本）。它允许将记录流写入一个或多个Kafka主题。

上面的示例演示了创建Flink Kafka Producer以便将流写入单个Kafka目标主题的基本用法。对于更高级的用法，还有其他构造函数变体可以提供以下功能：

默认情况下，如果未为Flink Kafka Producer指定自定义分区程序，则生产者将使用一个FlinkFixedPartitioner将每个Flink Kafka Producer并行子任务映射到单个Kafka分区（即，接收器子任务接收到的所有记录都将以相同的结尾Kafka分区）。

可以通过扩展FlinkKafkaPartitioner类来实现自定义分区程序。所有Kafka版本的构造函数都允许在实例化生产者时提供自定义分区程序。请注意，分区器实现必须是可序列化的，因为它们将在Flink节点之间传输。另外，请记住，由于分区不属于生产者检查点状态的一部分，因此在作业失败时分区器中的任何状态都将丢失。

也可以完全避免使用某种分区程序，而只需让Kafka通过它们的附加键（使用提供的序列化模式为每条记录确定）对书面记录进行分区即可。为此，请在实例化生产者时提供自定义分区程序。提供null作为自定义分区很重要；如上所述，如果未指定自定义分区程序，FlinkFixedPartitioner则使用。

Kafka 0.8 在0.9之前，Kafka没有提供任何机制来保证至少一次或完全一次语义。

Kafka 0.9和0.10 启用Flink的检查点后，FlinkKafkaProducer09和FlinkKafkaProducer010 可以提供至少一次的交付保证。

除了使用Flink的检查点，你还应该适当地配置setter方法setLogFailuresOnly(boolean)和setFlushOnCheckpoint(boolean)。

总之，0.9和0.10版本Kafka生产者在默认情况下具有-至少一次保证，与setLogFailureOnly设置为false和setFlushOnCheckpoint设置为true一样。

注意：默认情况下，重试次数设置为“ 0”。这意味着当setLogFailuresOnly设置为时false，生产者将立即因包括领导者变更在内的错误而失败。默认情况下，该值设置为“ 0”，以避免在目标主题中由重试引起的重复消息。对于大多数频繁更改代理的生产环境，建议将重试次数设置为较高的值。

注意：目前尚无Kafka的事务处理生产者，因此Flink无法保证一次准确地将消息发送到Kafka主题中。

Kafka 0.11及更高版本 启用Flink的检查点后，FlinkKafkaProducer011（FlinkKafkaProducer对于Kafka> = 1.0.0版本）可以提供一次准确的交付保证。

除了启用Flink的检查点之外，还可以通过将适当的semantic参数传递给FlinkKafkaProducer011（FlinkKafkaProducer对于Kafka> = 1.0.0版本）来选择三种不同的操作模式：

注意事项 Semantic.EXACTLY_ONCE模式依赖于提交从所述检查点恢复之后在采取检查点之前启动的事务的能力。如果Flink应用程序崩溃到完成重启之间的时间大于Kafka的事务超时时间，则将丢失数据（Kafka将自动中止超过超时时间的事务）。请记住，请根据预期停机时间适当配置事务超时。

Kafka Producer默认transaction.max.timeout.ms设置为15分钟。此属性不允许为生产者设置大于其值的事务超时。 FlinkKafkaProducer011默认情况下，将transaction.timeout.ms生产者配置中的属性设置为1小时，因此transaction.max.timeout.ms在使用该Semantic.EXACTLY_ONCE模式之前应增加 该属性。

在KafkaConsumer的read_committed模式下，任何未完成（未中止或未完成）的事务将阻止所有未完成的事务从给定的Kafka主题进行的所有读取。换句话说，在以下事件序列之后：

即使来自transaction2的记录已经提交，在transaction1提交或中止之前，消费者也看不到它们。这有两个含义：

注意：请Semantic.EXACTLY_ONCE采取一切可能的措施，以免留下任何为处理的transactions，而这将有必要阻止消费者继续阅读Kafka主题。但是，如果Flink应用程序在第一个检查点之前失败，则在重新启动此类应用程序后，系统中将不会包含有关先前池大小的信息。因此，FlinkKafkaProducer011.SAFE_SCALE_DOWN_FACTOR在第一个检查点完成之前按比例缩小大于Flink的比例来缩小Flink应用程序是不安全的。

从Apache Kafka 0.10+开始，Kafka的消息可以带有时间戳，指示事件发生的时间或消息已被写入Kafka代理的时间。

在FlinkKafkaConsumer010将消费带有时间戳的记录，如果在Flink时间特性被设定为TimeCharacteristic.EventTime(StreamExecutionEnvironment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime))。

Kafka使用者不会发出水印。为了发出水印，适用与上述使用该assignTimestampsAndWatermarks方法的“Kafka消费者和时间戳提取/水印发射”中所述的相同机制。

使用Kafka的时间戳时，无需定义时间戳提取器。该方法的previousElementTimestamp参数extractTimestamp()包含Kafka消息携带的时间戳。

Kafka消费者的时间戳提取器如下所示：

该FlinkKafkaProducer010只发出了纪录的时间戳，使用setWriteTimestampToKafka(true)设置。

Flink的Kafka连接器通过Flink的指标系统提供了一些指标，以分析连接器的行为。生产者通过Flink的度量标准系统为所有受支持的版本导出Kafka的内部度量标准。使用者导出从Kafka 0.9版开始的所有指标。Kafka文档在其文档中列出了所有导出的指标。

除了这些指标之外，所有消费者公开针对每个主题分区的current-offsets和committed-offsets。current-offsets是指当前分区中的偏移量。这是指我们成功检索并发出的最后一个元素的偏移量。committed-offsets是最后提交的偏移量。

Flink中的Kafka消费者将offsets提交给Zookeeper（Kafka 0.8）或Kafka broker（Kafka 0.9+）。如果禁用了检查点，则将定期提交偏移量。使用检查点时，一旦流拓扑中的所有操作都确认已创建其状态的检查点，就会进行提交。这为用户提供了至少一次提交给Zookeeper或代理的偏移量的语义。对于指向Flink的偏移量，系统仅提供一次保证。

提交给ZK或代理的偏移量也可以用于跟踪Kafka消费者的读取进度。每个分区中提交的偏移量和最新偏移量之间的差称为使用者延迟。如果Flink拓扑使用主题的数据的速度比添加新数据的速度慢，则延迟将增加，并且消费方将落后。对于大型生产部署，我们建议监视该指标以避免增加延迟。

Flink通过Kafka连接器提供一流的支持，以验证为Kerberos配置的Kafka安装。只需配置Flink flink-conf.yaml即可为Kafka启用Kerberos身份验证，如下所示： 1.通过设置以下内容配置Kerberos凭据：

2.附加KafkaClient到security.kerberos.login.contexts：告诉Flink向Kafka登录上下文提供配置的Kerberos凭据，以用于Kafka身份验证。

一旦启用了基于Kerberos的Flink安全性，就可以通过Flink Kafka消费者或生产者向Kafka进行身份验证，只需在传递给内部Kafka客户端的提供的属性配置中简单地包含以下两个设置即可：