同步Kafka生产者有例外吗？

文章/答案/技术大牛

发布

1回答

apache-kafka、confluent-platform、strimzi

正如您所知道的，有两种发送消息的方式-同步和异步。当我们使用同步模式编写代码时，代码如下所示 producer.send(new ProducerRecord(topicName, event)).get(); 阅读Kafka文档https://kafka.apache.org/10/javadoc/org/apache/kafka/clients/producer/KafkaProduc

浏览 29提问于2021-02-25得票数 1

回答已采纳

1回答

Scala中的Kafka 0.8.1异步生成器-无法检测到Kafka群集已关闭

scala、apache-kafka

我有一个非常奇怪的问题，我在Scala中实现了一个Kafka生产者客户端，当Kafka集群启动时，一切都正常工作。也就是说，生产者将数据正确推送到Kafka中。然而，如果我设置我的生产者是‘异步’，Kafka生产者客户端将正确地放置数据，如果集群启动和工作，但生产者也推送数据into...the无效...如果集群关闭了！也就是说，如果集群宕机，Kafka生产者不会返回错误。这对我

浏览 2提问于2014-10-29得票数 0

1回答

当存在不同步的副本和ACK=-1时，record会承诺给Kafka吗？

apache-kafka

运行合流- Kafka 3.3.1e Kafka 0.11.0x 2)如何使不同步的副本恢复同步？

浏览 0提问于2019-03-20得票数 0

回答已采纳

3回答

卡夫卡生产者行为

apache-kafka、kafka-producer-api

我在我的本地主机上设置了Kafka，并试图监控Kafka生产者的行为，以防出现网络问题。即使当所有代理都关闭时，生产者也不会给出任何错误。我使用的是同步制作器和Kafka 0.8版。如果所有的代理都宕机了，生产者有没有办法收到异常？

浏览 0提问于2015-10-16得票数 0

2回答

测试Kafka和was，NetworkException:服务器在收到响应之前断开连接

apache-kafka

运行凝聚式Kafka 4.1.1社区。所有其他生产者设置都是默认的。我启动了我的应用程序，当它开始向Kafka 写记录时，我就故意把其中一位经纪人解雇了，和我立即得到了：org.apache.kafka.common.errors.NetworkException: The生产者不应该写()成功，而不是抛出一个错误吗

浏览 0提问于2019-04-04得票数 3

回答已采纳

2回答

使Kafka生产者和消费者同步

apache-kafka

我有一个卡夫卡制作人和consumer.The卡夫卡制作人发布到一个主题，数据被获取和一些处理完成。kafka消费者正在从另一个主题中读取关于主题1中的数据处理是否成功或失败的信息，即主题2是否成功或失败。我正在启动我的消费者，然后将数据发布到主题1。.I想使生产者和消费者同步，即一旦生产者发布数据，消费者应该读取该数据的成功或失败消息，然后生产者应该继续处理下一组数据。

浏览 3提问于2017-08-01得票数 2

1回答

Kafka rest代理是解决方案吗？

rest、proxy、apache-kafka

流程：前端(img) -> ApiRest (Spring Boot) ->生产者->消费者(Python) ->向客户端返回响应。我知道我需要Kafka作为一个请求-响应，将响应返回给发出事件的客户端。我找对地方了吗？另一方面，我想知道在同一主题中使用不同语言(Java和Python)的几个使用者是否有任何问题。

浏览 26提问于2019-04-08得票数 0

1回答

Kafka Producer -默认支持多线程？

apache-kafka

我已经使用kafka_2.9.2-0.8.1.1创建了示例kafka同步生产者和消费者组程序。所以我的问题是，我是否需要向生产者添加多线程代码(就像consumergroup类一样)来支持大量的请求？我读到生产者发送方法是线程安全的。那么kafka producer将在内部处理多线程概念，还是开发人员必须显式编码？谢谢，Cdhar

浏览 2提问于2015-02-16得票数 2

1回答

卡夫卡消费者集团挂牌

java、apache-kafka

配置详细信息：(a)在VMPlayer Ubuntu操作系统中运行；(b)运行kafka版本: 0.8.0；(c)运行相同VM上的动物园管理员、Kafka、使用者和生产者(ubuntu/localhost)；(d)使用与此版本的kafka捆绑在一起的动物园管理员；(e)运行生产者示例(未更改，可在cwiki.apache.org/confluence/display/kafka/)获得；(f)运行消费群示例(未更改，ConsumerGro

浏览 1提问于2014-01-19得票数 4

回答已采纳

1回答

合流Rest代理API (生产者)重试事件是否在失败时发布？

rest、events、proxy、pipeline、confluent-platform

我们计划使用汇合Rest代理作为平台层，用于侦听所有用户事件(并将它们发布到Kafka)。使用微服务模型&具有不同类型的事件生成器，我们希望我们的API/事件生成器与事件侦听/处理分离。据我所知，如果Rest代理层由于任何原因未能发布到Kafka(一次)，则不会重试。此功能需要由调用方(客户层)处理，它需要进行同步调用&在失败时重试。但是，无法在产品文档中找到有关这方面的任何细节。谁能确认一下同样的情况吗？合流Rest代理开发人员声称，通过客户端设置正确的Rest代理集群和正确的请

浏览 5提问于2017-09-14得票数 0

1回答

我可以用多少制作者来写一个主题？

apache-kafka

我有一个网络应用程序，它把信息放入一个卡夫卡主题。这个应用程序有很多实例(200)，每个实例都包含它自己的Kafka生产者。是否存在每个主题的生产者数量上限？生产者的最佳做法是什么？每个应用程序都有一个同步生成器，一个异步生成器，还是一个自定义的同步生成器池？

浏览 2提问于2015-05-06得票数 5

回答已采纳

2回答

如何在卡夫卡java中生成所有消息后发送元数据给主题？

java、apache-kafka、kafka-producer-api

我正在使用Java Kafka客户端库为Kafka主题制作学生详细信息，我想要的是，一旦所有记录都生成，我想将消息计数与其他一些元数据一起发送到同一个Kafka主题。

浏览 16提问于2022-07-01得票数 0

1回答

部分Kafka节点时间偏移量不同步导致Spark流作业卡住

java、apache-spark、apache-kafka、spark-streaming、kafka-producer-api

我们有一个spark streaming作业，它从Gnip API读取并向Kafka集群发送tweet。 Kafka集群是使用Cloudera Manager安装的。有时，cloudera管理器会显示一些Kafka节点的健康状况不佳的消息。不良运行状况消息与NTP服务相关。某些节点突然与NTP服务器不同步。为什么Kafka节点和NTP服务器同步会影响spark流作业中的Kafka生产者？

浏览 1提问于2016-09-04得票数 4

2回答

卡夫卡卡夫卡是按同样顺序收到的信息吗？

java、apache-kafka、kafka-producer-api

我正在处理一个从IBM收集数据并将其处理为kafka主题的过程。为了确保不丢失任何消息，我需要在确保kafka broker发送和接收消息之后才提交JMS消息。我不想使用同步卡夫卡生产者(等待future.get())，因为它可能会对性能产生影响，相反，我想在我提供的kafka生产者提供的回调中提交我的JMS消息。我的假设正确吗？

浏览 12提问于2022-05-10得票数 0

2回答

有办法在Spring数据流中配置Kafka客户端超时设置吗？

spring-cloud-stream、spring-cloud-dataflow

在启动Spring数据流流时，由于与Kafka相关的各种错误，应用程序常常无法在我的机器上部署。Caused by: kafka.common.KafkaException: fetching topic metadata for topics [Set(xxx)] from broker [List()] failed 但我也看到了这个：kafka.admin.AdminOperationException: replication factor: 1 larger than available有什么办法可以影响卡夫卡客

浏览 2提问于2016-10-04得票数 2

回答已采纳

1回答

Spring云流MessageChannel send()总是返回true

spring、spring-boot、spring-cloud-stream、spring-kafka

我使用stream，我想保存消息，并在Kafka服务器离开时重试发布这些消息，但是MessageChannel send()方法总是返回true，即使停止了Kafka/Zookeeper服务器。有人能帮忙吗？具有内容的application.yml更新: cloud: kafka: binder:source.output().send(MessageBuilder.withPayload(notif)

浏览 4提问于2017-12-06得票数 5

4回答