apache-kafka 相关问题

“acks”配置是否适用于 KafkaProducer 客户端的“sender”线程？

我有几个关于 KafkaProducer 客户端的异步性质的问题，官方文档没有回答这些问题。 背景： 我正在做一个项目，我需要记录...

KStreams 外连接的作用类似于内连接

我正在使用 .outerJoin 将两个流连接在一起。预期的行为是我会在输出中获得两个输入中所有记录的记录；但实际上我只得到一个输出，其中 t...

有什么方法可以让我们在Java Enterprice应用程序中启动kafkabroker和zookeeper，而无需使用命令提示符手动启动？

我想知道启动kafkabroker和zookeeper的编程方式。 我需要指导，我尝试使用这些 bewlow 命令手动启动 kafka 代理和 kafka 动物园管理员， 开始...

kafka：取消了正在进行的 METADATA 请求，由于请求超时而与节点断开连接

我有 spring-kafka:3.0.3、SSL，我能够将我的消费者连接到引导服务器并成功消费消息。 但除此之外我看到这样的日志消息 飞行途中取消

我可以在 AWS MSK 上以 KRaft 模式使用 Kafka 吗？

我想在KRaft模式下使用Kafka而不是AWS上的Zookeeper，因为Kraft模式似乎是Kafka的未来。据我了解，AWS针对Kafka的基本解决方案是AWS MSK。但在 MSK

如何控制我的 kafka 流应用程序中用于 leftJoin 操作的内部主题的名称？

我正在一个环境中构建一个kafka流应用程序，该应用程序不允许具有创建内部主题的管理员权限。我可以通过让我的操作具有

在 Amazon MSK（公共 Kafka）上创建主题授权问题

我们有一个在 Amazon AWS 上运行的 Kafka 服务器。这是启用了 SASL/SCRAM 的公共服务器。我们可以成功进行身份验证，但是当我们尝试创建新主题时，我们收到授权错误......

Quarkus + Kafka 拦截传入消息并读取标头

有没有办法使用quarkus中的smallrye连接器为来自Kafka的传入消息创建拦截器？我需要它来读取标头并切换租户以保存数据。 今年四月...

Kafka 状态存储跨子拓扑共享

我正在尝试创建一个自定义加入消费者来加入多个事件。 我创建了一个拓扑，它有四个子拓扑（subtopology-0、subtopology-1、subtopology-2、subtopology-3），不在

具有KRAFT认证的Kafka

我有 kafka 和 KRAFT，我希望我的用户（在 python 中）只能访问专门为他们创建的特定主题。 请任何建议，我完全不明白如何制作。我可以

Kafka 消费者注册表每次保存都需要很长时间

我对kafka很陌生，我有一个init函数来设置生产者、消费者并订阅我拥有的所有主题。目前，我们的每个 Express 端点（res、req）有 20 个主题。它

从kafka读取数据并写入hadoop/hbase/hive/spark堆栈批量处理

我有一个技术堆栈，其中 Hadoop 作为原始数据的分布式存储，HBase 作为在 Hadoop 和 HDFS 之上运行的 NoSQL 数据库，Hive 作为 Hive 之上的 RDBMS 数据仓库，用于应用程序...

Quarkus 消息传递 - 验证连接器传出消息

我们使用 Quarkus 来消费来自 Kafka 的事件，执行一些丰富操作，并使用自定义连接器通过 HTTP 将它们推送到内部系统。 我们需要验证丰富的事件是否...

Openshift 到 Kafka 服务器的路由

是否可以创建到Openshift中部署的Kafka服务器的Openshift路由，以便使用Kafka工具从本地计算机连接Kafka？ 有一个 Pod（Kafka、Zookeeper）， 服务（

Zookeeper f-sync 问题导致 Kafka 超时

我们正在尝试在具有 16 核和 16 GB RAM 的 VM 上为 Zookeeper 和 Kafka 进行单节点安装。 Zookeeper 显示 f-sync 问题，并且在设置的 6000 毫秒超时期限之后 kafka gi...

消费者中的@KafkaHandler不会将主题消息作为对象类消费，而仅作为字符串消费

我有一个简单的 Java Spring Boot 应用程序，它应该将来自 Kafka 的主题消息作为对象类使用，但它只能作为字符串调用。仅调用了handleDefault..为什么？ @

Kafka - 重置分区偏移量不起作用

我有一个主题“oranges”，有 10 个分区，1 个消费者组内有 2 个消费者。我正在使用 Spring Kafka。 由于某些原因，我需要时不时地重新读取数据，我需要重置

当我使用 Apache Kafka 重新启动应用程序时，我想再次使用相同的消息，但这需要几乎 1 分钟

我有一个控制台应用程序（消费者）。当我重新运行我的应用程序时，我想再次使用相同的消息。我已经设置：EnableAutoCommit = false 但每次我重新运行我的应用程序时......

Spark消费者使用docker运行时找不到kafka主题分区

当我提交连接到 kafka 代理的 Spark 应用程序时，它会执行 kafka 查询，但不会将任何内容返回到控制台。找不到主题分区。 这是我的日志

Spring Cloud Kafka Binder 设置指定消费者组的自动偏移重置不会更改默认值

我正在尝试从最新的可用偏移量中读取（没有为消费者组提交偏移量），所以我在属性中指定 spring.cloud.stream.bindings.process-in-0.group=测试组 spring.cloud.stream.