《kafka》专题

当使用int-kafka： out站通道适配器生成到kafka时，似乎没有可用的错误通道。在这种情况下，如何处理reties次后无法向kafka生成消息？ 任何可能导致kafka失败的错误。（以下代码只是来自Internet的代码片段，只是想知道如何向其添加错误句柄）

我正在使用带有以下配置的spring集成kafka 1.1.0。我不太了解流的配置。当我增加这个值时，Spring会自动生成更多线程来处理消息吗？e、 g.当我有流=2时，相关的转换器和服务激活器是否都在2个线程中运行？我想错过一些线程执行器配置，但不知道如何配置。任何提示都将不胜感激。谢谢 已尝试使用消息驱动通道适配器，但无法使其工作，以下配置未拾取任何消息。还尝试了组织。springframe

我试图在spring集成中实现一个自定义入站通道适配器，以使用来自apache kafka的消息。基于spring集成示例，我发现需要创建一个实现MessageSource接口的类，并实现receive（）方法，该方法将返回来自kafka的已使用消息。但根据kafka中的消费者示例，KafkaStream中的消息迭代器由BlockingQueue支持。因此，如果队列中没有消息，线程将被阻塞。 那么

我正在开发一个spring集成应用程序，它有一个Kafka出站通道适配器，并使用spring集成java dsl配置流。 spring集成核心：4.2.4。释放 spring集成Kafka：1.3.0。释放 spring集成java dsl：1.1.2。发布 我已经配置了类似于以下代码段的消息处理程序规范。 我想添加一个ProducerListener。此功能已添加到此处的Spring集成kafk

我开发了一个Spring批处理作业，它使用KafkaItemReader类读取Kafka主题。我只想在处理在定义块中读取的消息并将其成功写入输出时提交偏移量。dat文件。 ============================================================================== ===================================

我在本地计算机上将一条太大的消息推送到Kafka消息主题中，现在我收到一个错误： 增加在这里并不理想，因为我实际上不想接受那么大的消息。

我有一个需要使用Kafka Console Producer发送键值消息的用例。那么如何通过命令实现这一点呢？

我想设计一个解决方案，用于向多个提供商发送不同类型的电子邮件。总体概述。 我有几个上游提供商Sendgrid、Zoho、Mailgun等。它们将用于发送电子邮件等。例如： 注册新用户的电子邮件 删除用户的电子邮件 空间配额限制的电子邮件 （一般大约有6种类型的电子邮件） 每种类型的电子邮件都应该生成到生产者中，转换为序列化Java对象，并发送到与上游提供商集成的适当的Kafka消费者。 问题是如何

嗨，我正在阅读kafka主题，我想处理从kafka接收到的数据，例如tockenize，过滤掉不必要的数据，删除停用词，最后我想写回另一个kafka主题 然后我得到以下错误 线程"main"中的异常org.apache.spark.sql.Analysis Exception：具有流源的查询必须使用WriteStream.start（）执行； 然后，我对代码进行了如下编辑，以从Kafka中读取并写

我是Kafka生态系统的新手，在我的例子中，我使用的是Java生产者，但不需要发送密钥以及序列化Avro的记录值。有没有办法构建一个Java生产者来不发送密钥，或者在Kafka中发送消息时需要密钥？

我遇到了两个关于订购的短语， 生产者发送到特定主题分区的消息将按发送顺序追加。也就是说，如果记录M1与记录M2由同一生产者发送，并且M1首先发送，则M1的偏移量将低于M2并出现在日志中的较早位置。 另一个 问题是，如果存在如#2所述的失败发送，那么该顺序是否仍会保留到特定分区？如果一条消息存在潜在问题，将删除每个分区的所有以下消息“以保留顺序”，或者将发送“正确”的消息，并将失败的消息通知应用程序

如何在拆分后将所有消息放入Kafka后执行方法。 我尝试在频道后使用句柄（），但最终没有从第2条消息发布到Kafka。 我想在将所有拆分消息写入Kafka后执行一个方法。

我试图发布一些消息使用Kafka消费者与"回复Kafka模板"。我的主要工作是订阅消息，修改消息，并发回修改后的消息。我已经尝试了增加回复时间。但即使如此，我也没有得到订户的回应。生产者控制台显示如下。 我已尝试增加事务超时、请求超时。但对我来说什么都不管用。任何帮助都将不胜感激。 提前谢谢 这些是我的配置bean： 这是我的消费者： 警告8088---[TaskScheduler-1]o.s.k

我正在运行一个spring boot应用程序，它会侦听IBM消息中心Kafka，然后将其存储到IBM云上的compose for mongoDB中。 在一起运行Kafka和Compose for MongoDB时，我遇到了一个与SSL证书相关的错误。 我正在为mongoDB证书使用compose并使用有效的信任存储和密钥。 如果我在Spring靴中使用嵌入式mongo，那么一切都很好。 到目前为止

我试图连接到Kafka，从C#client启用TLS，在调用。不幸的是，到目前为止，互联网上的帖子都没有帮助我解决这个问题。你知道会出什么问题吗？ 下面是我用来启动连接的最小样本C#代码