《kafka》专题

在我的Spring启动kafka发布者应用程序中，我希望支持以字符串（json）或字节格式发布消息，因为我希望同时支持json和avro。但spring boot中的Kafka模板让我们只定义其中一个模板。有没有一种方法可以同时使用两个模板或任何其他方式来提供对json和avro的支持？

使用Spring Integration Kafka，使用出站通道适配器，我尝试向名为“test”的主题发送消息 通过命令行终端，我启动了动物园管理员、kafka并创建了名为“test”的主题 Spring XML配置 JUnit测试代码 测试用例成功，在调试时，我发现channel.send()返回true 我使用下面的命令通过命令行检查了主题，但是我在测试主题中看不到任何消息。 bin/kaf

当一个DLQ被设置为一个Spring云流Kafka消费者时，DLQ写入的主题可以被分区吗？我有一个要求，使密钥等于一个特定的字段，我想知道这将如何与Spring云流。

我正在尝试使用Spring云Kafka流绑定器来使用来自主题的Avro消息，但无法修复此class Cast异常。 这是我的代码： 粘合剂: 错误： 我尝试了本链接中提到的两种方法https://spring.io/blog/2019/12/04/stream-processing-with-spring-cloud-stream-and-apache-kafka-streams-part-3-d

我有一个场景，比如。 有4个主题处理主题，异常主题，重试主题和拒绝主题。我有一个spring cloud stream应用程序，它有一个使用Kstream的处理器。该处理器从异常主题中读取消息，并基于每个消息中可用的标志，为重试主题和拒绝主题创建两个kstream分支。现在需要做的是，retry主题中出现的任何消息都必须等待一段特定的时间，然后才能将其推回处理主题。有没有人能帮我在spring c

我正在尝试弄清楚如何测试我的Spring Cloud Streams Kafka-Streams应用程序。 应用程序如下所示： 流1：主题1 我尝试了不同的方法，例如TestChannelBinder，但这种方法仅适用于简单函数，而不适用于Streams和Avro。 我决定将EmbeddedKafka与MockSchemaRegistryClient一起使用。我可以生成一个主题，也可以再次使用同一

使用Spring Cloud DataFlow 1.2.2版本，配置如下： 我正在尝试创建一个流，它将从特定主题中读取并将其刷新到长水槽中，如下所示： 查看日志文件，我可以看到以下错误： 我还试图为kafka源代码的消费者/生产者配置一些属性 但我得到的结果是一样的 以下是Spring DataFlow打印的消费者详细信息: 我看到了类似的查询，但没有有效的答案，什么是属性来接受二进制json消息

我们刚刚升级到Spring-Cloud-Stream的3 . 0 . 0-版本，遇到了以下问题: 像这样使用函数式风格时: application.yaml如下所示： 似乎序列化做了两次——当我们截取kafka主题中产生的消息时，消费者只是将它们显示为JSON(字符串),但现在它是一个不可读的字节[]。此外，生产中的下游消费者不能再反序列化消息。奇怪的是，输入消息的反序列化似乎工作得很好，无论我们

我正在使用Spring Cloud Stream Kafka Binder。我有以下Kafka活页夹函数。 在yml中，我有: 如果我想从同一个功能向两个不同的主题发送数据，我需要做什么？

我正在迁移一个Kafka Streams实现，它使用纯Kafka apis来使用sping-kafka，因为它被合并在sping-引导应用程序中。 一切都很好Stream，GlobalKtable，分支，我所有的工作都非常好，但我很难合并ReadOnlyKeyValueStore。基于这里的sping-kafka留档：https://docs.spring.io/spring-kafka/docs

以下是我的情况： 我们有一个Spring cloud Stream 3 Kafka服务连接到同一个代理中的多个主题，但我想基于属性控制连接到特定主题。 每个主题都有自己的活页夹和绑定，但代理对所有人来说都是一样的。 我尝试使用下面的属性禁用绑定（这是我到目前为止找到的唯一解决方案），这适用于StreamListener不接收消息，但与主题的连接和重新平衡仍在发生。 我想知道活页夹级别是否有任何设置

如何使用来自 Kerberos 授权的 kafka（版本 0.10）服务器的已发布消息，因为正在使用身份验证密钥表文件。 我尝试使用以下命令，但没有显示输出。 bin/kafka-console-consumer.sh --bootstrap-server ：9092 --topic --from-start

我已经建立了一个由3个节点组成的AWS集群。我修改了节点的/etc/hosts文件，看起来像这样 当我从其中一个节点运行命令时 bin/kafka-console-consumer.sh --bootstrap-server kafka1：9092 --topic first_topic --from-start 它可以工作，但是当我用ip替换主机名并用下面的命令运行它时 bin/kafka-co

我想使用以下命令使用来自 Kafka 主题的数据，如下所示： kafka-console-consumer.sh引导服务器localhost:9092 然后，这将输出以下内容(只粘贴前2行输出，但它将是许多行...): 我想使用输出中的以下内容: > 房地产经纪人 家庭标识 消息 唯一标识 您可以看到，每个元素都用一个&符号（' 描述测试表； SELECT*FROM测试表； 到目前为止，我有什么？

我是春云和Kafka流的新手。我正在尝试使用 kafka 活页夹设置Spring云应用程序。我尝试在本地测试 kafka 流处理器，但我无法打印任何日志。 我的kafka消息将包含JSONObject。kafkaStreamListener类是： Application.properties： 问题:在调试模式下，断点直接到达过滤器步骤，然后不执行任何操作。它跳过了记录器和SOP。不知道可能是什么