程序员经验分享-hwhale

卡夫卡消费者寻求开始

2024-01-03

我没有使用分区来发布到 Kafka 主题。 ProducerRecord（字符串主题，K键，V值）

在消费者方面，我想从头开始。 eekToBeginning（集合分区）

是否可以在不使用分区的情况下寻求开始？ Kafka 是否分配默认分区？

https://kafka.apache.org/0102/javadoc/org/apache/kafka/clients/ Producer/ProducerRecord.html https://kafka.apache.org/0102/javadoc/org/apache/kafka/clients/producer/ProducerRecord.html https://kafka.apache.org/0102/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html https://kafka.apache.org/0102/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html


在生产时，如果您没有明确指定分区，生产者会自动从您的主题中选择一个分区。

在您的消费者中，如果您订阅了您的主题，您可以使用以下方式查找您的消费者当前分配给的所有分区的开头：

consumer.seekToBeginning(consumer.assignment())
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

ApacheKafka

kafkaconsumerapi

卡夫卡消费者寻求开始 的相关文章

  • KafkaConsumer.commitAsync() 行为的偏移量比以前更低

    kafka 将如何处理调用 KafkaConsumer commitAsync Map

  • Apache Kafka 是否提供异步订阅回调 API？

    我的项目正在将 Apache Kafka 视为老化的基于 JMS 的消息传递方法的潜在替代品 为了让这个过渡尽可能的顺利 如果替代的排队系统 Kafka 有一个异步订阅机制那就更理想了 类似于我们当前项目使用的JMS机制MessageLis

  • kafka ProducerRecord 和 KeyedMessage 有什么区别

    我正在衡量卡夫卡生产者生产者的表现 目前我遇到了两个配置和用法略有不同的客户 Common def buildKafkaConfig hosts String port Int Properties val props new Proper

  • 从 Apache Kafka 中的主题删除消息

    所以我是 Apache Kafka 的新手 我正在尝试创建一个简单的应用程序 以便我可以更好地理解 API 我知道这个问题在这里被问了很多 但是如何清除存储在主题上的消息 记录 我看到的大多数答案都说要更改消息保留时间或删除并重新创建主题

  • kafka消费者群体正在重新平衡

    我正在使用 Kafka 9 和新的 java 消费者 我正在循环内进行轮询 当代码尝试执行 Consumer commitSycn 时 由于组重新平衡 我收到 commitfailedexcption 请注意 我将 session time

  • 有没有办法使用 .NET 中的 Kafka Ksql Push 查询

    我目前正在 NET 中使用 Kafka 消费者处理大量 Kafka 消息 我的处理过程的第一步是解析 JSON 并根据 JSON 中特定字段的值丢弃许多消息 我不想首先处理 特别是不下载 那些不需要的消息 看起来 kSql 查询 写为推送查

  • 未能在kafka-storm中将偏移量数据写入zookeeper

    我正在设置一个风暴集群来计算实时趋势和其他统计数据 但是我在将 恢复 功能引入到这个项目中时遇到了一些问题 方法是允许上次读取的偏移量kafka spout 源代码为kafka spout来自https github com apache

  • 如何在 PySpark 中使用 foreach 或 foreachBatch 写入数据库？

    我想使用 Python PySpark 从 Kafka 源到 MariaDB 进行 Spark 结构化流处理 Spark 2 4 x 我想使用流式 Spark 数据帧 而不是静态数据帧或 Pandas 数据帧 看来必须要用foreach o

  • Kafka 中的内部和外部通信

    流动 本地 gt 代理 gt Kafka advertised listeners PLAINTEXT proxyhostname 8080 for external communication listeners PLAINTEXT 90

  • 如何使用 Python 在 Kafka 中生成 Tombstone Avro 记录？

    我的水槽属性 name jdbc oracle config connector class io confluent connect jdbc JdbcSinkConnector tasks max 1 topics orders con

  • 带有 spring-kafka 的 Kafka 死信队列 (DLQ)

    最好的实施方式是什么死信队列 DLQ Spring Boot 2 0 应用程序中的概念 使用 spring kafka 2 1 x 来处理无法处理的所有消息 KafkaListener某些bean发送到某些预定义的Kafka DLQ主题的方

  • Kafka Streams 反序列化处理程序

    我正在尝试在反序列化中使用 LogAndContinueExceptionHandler 当发生错误时 通过成功记录错误并继续 它可以正常工作 但是 假设我的传入消息有连续的错误流 我停止并重新启动 kafka 流应用程序 然后我看到失败并

  • Kafka JDBC Sink Connector，批量插入值

    我每秒收到很多消息 通过 http 协议 50000 100000 并希望将它们保存到 PostgreSql 我决定使用 Kafka JDBC Sink 来实现此目的 消息以一条记录保存到数据库 而不是批量保存 我想在 PostgreSQL

  • 即使在 Kafka 中进行轮询后，当前也不会发生分区分配

    我有 Java 8 应用程序与 Apache Kafka 2 11 0 10 1 0 一起使用 我需要使用seek特征为poll来自分区的旧消息 然而我遇到了一个例外No current assignment for partition每次

  • Apache Kafka 中消费者消费消息的延迟

    我正在使用 Kafka 0 8 0 并尝试实现下面提到的场景 JCA API 充当生产者并将数据发送到 gt 消费者 gt HBase 一旦我使用 JCA 客户端获取数据 我就会将每条消息发送给消费者 例如 一旦生产者发送消息 no 1 我

  • 是否有用于事件驱动的 Kafka 消费者的 Python API？

    我一直在尝试构建一个以 Kafka 作为唯一界面的 Flask 应用程序 因此 我希望有一个 Kafka 消费者 当相关主题的流中存在新消息时 该消费者会被触发 并通过将消息推回到 Kafka 流来进行响应 我一直在寻找类似 Spring

  • 提供了 kafka schema.registry.url 但不是已知的配置

    尝试使用架构注册表发布有关主题的 json 消息 但出现以下错误 以下Spring Boot方法 已提供配置 schema registry url 但不是已知配置 应用程序 yml 文件 server port 9080 spring k

  • 如何在kafka中定义多个序列化器？

    比如说 我发布和使用不同类型的 java 对象 对于每个对象 我必须定义自己的序列化器实现 我们如何在 serializer class 属性下提供kafka消费者 生产者属性文件中的所有实现 我们有一个类似的设置 不同主题中的不同对象 但

  • Windows 上的 Apache Kafka 错误 - 无法找到或加载主类 QuorumPeerMain

    我刚刚从 Apache 网站下载了 Kafka 2 8 0 我正在尝试使用网站上给出的说明进行设置 但是当我尝试启动 Zookeper 服务器时 出现以下错误 错误 无法找到或加载主类 org apache zookeeper server

  • 如何删除 Apache Kafka 中的多个主题

    假设我有许多具有相同前缀的主题 例如 giorgos topic1 giorgos topic2 giorgos topic3 用于删除单个主题的命令 例如giorgos topic1 如下 bin kafka topics sh zook

随机推荐

热门标签