《kafka》专题

我创建了一个python脚本raw\u tweets\u流。py使用twitter api流式传输twitter数据。twitter上的json数据使用下面的脚本传递给kafka producer。

D： \软件\Kafka\Kafka2.10-0.10.0.1\bin\windows 我使用上面的命令来消费消息，有什么我错过的吗？帮助我： 这个 那些是生产者和消费者......

我试图使用FlinkKafkaProducer010生成元素，但是当我打开消费者控制台窗口时，元素似乎出现了故障。 我使用kafka-topics.bat创建了主题 消费者是使用：kafka console consumer创建的。蝙蝠——zookeeper本地主机：2181——主题mytopic 我使用的Kafka制作人代码是： 当我查看Kafka日志文件时，我看到一个. log文件，其中的元素

我想在成功发送关于Kafka主题的所有消息后执行一些代码。我读了很多文章。要知道我们可以使用事务发送消息。 我尝试了这么短的代码，但不确定如何在这个事务成功执行后执行一些代码。我也尝试过Kafka普勒，开始交易，但这对我的案子也不起作用。如果有什么想法的话会很有帮助。

我的问题是，在使用kafka的spring boot微服务中，什么是适合使用KafkaTemplate的。send（）或KafkaProducer。发送（） 我使用了Kafka康苏默而不是Kafka利斯纳来调查记录，因为Kafka利斯纳正在获取记录，当他们谈到主题时，我希望根据业务需求定期调查记录。 已经看过Kafka制作人的文件了https://kafka.apache.org/10/javad

我正在为一个简单的Spring Boot应用程序编写Kafka联调。该应用程序简单地发布到Kafka主题。 我正在使用一个嵌入式Kafka实例进行测试。当通过Intellij运行时，测试运行得非常好，但当我通过gradle运行时，测试失败。看起来好像倒计时从未达到0，测试最终超时。 生产者配置： 制片人： 消费者： 测试： application.yml： 知道问题出在哪里吗？

Kafka新手。 Kafka版本：2.3.1 我正在尝试使用Spring cloud使用来自两个主题的Kafka消息。除了kafka活页夹和下面的一些简单配置之外，我没有做太多配置。每当（组协调器lbbb111a.uat.pncint.net:9092（id：2147483641机架：null）不可用或无效时，将尝试重新发现）发生时，已经处理的一堆消息会再次被处理。不确定发生了什么。

试图理解消费者补偿和消费者群体补偿之间的关系。 下面的堆栈溢出链接提供了对消费群体补偿管理的极好理解<什么决定Kafka消费补偿？现在问题来了， 情节： 我们在一个消费者组组1中有消费者（c1）。 偏移值是否将存储在消费者（c1）和组（group1）两个级别？或者如果消费者属于任何消费者组，偏移量将存储在仅消费者组级别？ 如果偏移值将存储在两个级别中，它是否是消费者级别偏移值将覆盖消费者组级别偏移

我对Spring的Kafka还是有点陌生。我的问题很简单。我有一个仅限消费者使用的应用程序，它可以连续读取Kafka，处理消息，并使用Ack侦听器手动确认消息。我有一个上游生产者专用应用程序的依赖项，在该应用程序中，他们负责向Kafka主题发送消息，以便我使用。我们最近在生产者和消费者之间实现了事务，但我想了解更多关于故障点的信息，以及如何处理那些回滚的事务，以便它们不会丢失？我已经读到，最好使用

一、 我正试图用Kafka信源和信宿测试Flink一次语义： 运行flink应用程序，只需将消息从一个主题传输到另一个主题，并行度=1，检查点间隔20秒 每2秒使用Python脚本生成具有递增整数的消息。 使用read_committed隔离级别的控制台使用者读取输出主题。 手动杀死TaskManager 我希望在输出主题中看到单调递增的整数，而不考虑TaskManager的终止和恢复。 但实际上

我正在使用Spring和Spring Kafka编写一个小型PoC。我的目标是让生产者和消费者都从这个主题中写作（例如阅读）。 我有一个奇怪的情况： 制作人正确地制作了这些记录（我可以通过Python脚本使用它们） 下面是我的代码——它与文档示例非常相似。更准确地说，问题来自这样一个事实，即KafkanConsumerConfiguration中的bean不是由Spring创建的（也就是说，构建它

我没有使用Spring Kafka模块来生成和使用消息。相反，我在生产者和消费者实现中使用Apache客户端库。由于我没有使用Spring Kafka，因此Spring Slueth自动配置不适用于生成跟踪。我已经提到https://docs.spring.io/spring-cloud-sleuth/docs/current-SNAPSHOT/reference/html/integration

我们有一个基于spring boot的事务性Kafka制作人！使用的版本如下 spring-boot-starter-父-2.3.0。释放 spring-kafka-2.5.0。释放 我们的kafka（集群）版本是2.1. x！ 作为生产者，我们启用了幂等性，定义了事务id前缀，并在事务中执行kafka模板调用。我们还有一个将隔离级别设置为只读的使用者！ 现在我们遇到了一个行为，不知道如何推断，

Kafka为每条消息生成偏移量。假设，我正在生成消息5，偏移量将从1到5。 但是，在事务生产者中，比如说，我产生了5条消息并提交，然后是5条消息但中止，然后是5条消息提交。 > 那么，最后提交的5条消息的偏移量是6到10还是11到15？ 如果我不放弃或不promise呢。这些信息还会被发布吗？ Kafka是如何忽略未promise的补偿的？因此，kafka提交日志是基于偏移量的。它是否使用事务使用

我想在我的spring boot项目中使用Kafka Streams实时处理。所以我需要Kafka Streams配置，或者我想使用KStreams或KTable，但我在互联网上找不到示例。 我做了制作人和消费者现在我想流实时。