是否可以让Kafka消费者/宿连接跳过topic内的特定分区？
Kafka消费者/宿连接跳过topic内的特定分区是不可行的。Kafka是一个分布式流处理平台，它通过将数据分成多个分区并在多个服务器上进行复制来实现高可用性和可扩展性。每个分区都是有序的，并且只能由一个消费者组中的一个消费者进行消费。
消费者组中的每个消费者负责消费一个或多个分区的数据。当消费者加入或离开消费者组时，Kafka会自动重新分配分区，以确保每个消费者负责的分区数量相对均衡。这种分区分配策略确保了数据的负载均衡和高吞吐量。
因此，如果一个消费者/宿连接跳过topic内的特定分区，那么这个分区的数据将无法被消费者处理，导致数据丢失或处理不完整。在Kafka中，消费者只能消费其分配到的分区数据，无法跳过或选择性地消费特定分区的数据。
对于需要跳过特定分区的需求，可以考虑使用Kafka Streams或Kafka Connect来进行数据处理和转换。Kafka Streams是一个用于构建实时流处理应用程序的客户端库，可以对数据进行过滤、转换和聚合等操作。Kafka Connect是一个用于可靠地将数据从外部系统导入和导出Kafka的工具，可以通过配置来选择性地导入或导出特定分区的数据。
腾讯云提供的相关产品是CKafka，它是腾讯云自研的分布式消息队列服务，基于Kafka协议构建。CKafka提供高可靠、高吞吐量的消息传输，适用于大规模数据流处理和实时数据分析等场景。您可以通过CKafka来实现数据的可靠传输和实时处理。
更多关于CKafka的信息和产品介绍，请访问腾讯云官方网站：CKafka产品介绍。
相关·内容
1回答
我已经使用python.How创建了实现的Kafka生产者-消费者消息传递主题我可以对队列做同样的事情，这样消息就只会发送给一个消费者。
这是我的生产者代码
# Import KafkaProducer from Kafka library
from kafka import KafkaProducer
# Define server with port
bootstrap_servers = ['localhost:9092']
# Define topic name where the message will publish
topicName = 'Firs
浏览 3提问于2020-12-03得票数 0
我的Kafka Producer正在以大约..的速度生成消息。每个30 seconds的350 mb。
Kafka Setup
--> 1 Zookeeper instance
--> 3 Kafka Brokers
--> 1 Java Producer
--> 1 Java Consumer
这就是我创建Topic and broker partitions的方式
bin/kafka-topics.sh --create --zookeeper 10.10.1.5:2181 --replication-factor 1 --partitions 8 --topic
浏览 1提问于2014-04-19得票数 1
1回答
我有一个具有多个主题的Kafka-Broker，每个主题都有一个分区。
我有一个消费者，它可以很好地使用主题中的消息
我的问题是，我需要通过增加分区的数量来提高消息队列的吞吐量，假设我在一个主题上有四个分区，是否有一种方法可以编写四个消费者，每个消费者指向该主题上的单个分区？
import java.util.*;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStre
浏览 25提问于2016-09-23得票数 4
回答已采纳
kafka使用者无法仅提交特定分区的偏移量。
aklsfoipafasldmaknfa asiofuasofiusaofasd
[2019-01-04 12:22:22,691] ERROR [Consumer clientId=consumer-1, groupId=console-consumer-11955] Offset commit failed on partition my-topic-2-9 at offset 0: The request timed out. (org.apache.kafka.clients.consumer.internals.ConsumerCo
浏览 3提问于2019-01-04得票数 5
我正在使用Kafka-go库构建一个请求-响应设置，使用消息键作为关联ID。我的设置在没有并发的情况下工作得很好，但是当消息开始在单独的goroutines中发送时，读取器部分会跳过正确的键(因为其他例程可能已经读取了它)。
考虑到连接由不同的goroutine共享，我如何从主题中只读取特定的键？
客户端示例如下(为简洁起见，删除了错误评估)：
package main
import (
"bytes"
"context"
"fmt"
"sync"
"time"
浏览 8提问于2019-07-27得票数 0
回答已采纳
我仍然不能清楚地理解并行性，比方说，我们有一个flink集群，它有足够的插槽。在我们的flink工作中，我们从3个不同的kafka集群中使用3个kafka主题，每个主题有10个分区。
如果我们想尽快使用该消息。并行性应该是什么？
如果我们将并行性设置为10，那么就会使用10个槽，这意味着，根据我的理解，10个线程对吗？如果这10个线程“连接”到topic_1，那么就没有线程读取topic_2和topic_3。
如果我们将并行性设置为30，那么就有30个线程，但这30个线程是否足够聪明，它们中的10个会转到topic_1，10个会转到topic_2，其余的10个会转到topic_3？
浏览 0提问于2019-03-13得票数 12
回答已采纳
我已经配置了Spring引导执行器和千分尺，用于公开Kafka消费者度量标准。卡夫卡的一些指标没有“主题”标签。例如,
kafka_consumer_records_consumed_total_records_total{client_id="consumer-1",topic="topic1",} 0.0
kafka_consumer_records_consumed_total_records_total{client_id="consumer-2",topic="topic2",} 0.0
kafka_consumer_r
浏览 4提问于2020-07-22得票数 0
1回答
我有一个kafka主题被划分成60个分区。生产者以默认的循环方式分发消息。到目前一切尚好。但在消费方面，滞后正在加剧。没什么好奇怪的。令我惊讶的是，这种滞后有时是如何分布的：
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
a_group a_topic 24 2013567
浏览 10提问于2022-09-27得票数 0
我在本地机器上设置了3个代理，我用5分区bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 3 --partitions 5 --topic test-partitions创建了一个主题
然后来描述我的test-partitions主题，
bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic test-partitions
这会导致
Topic: test-partitions Part
浏览 2提问于2020-05-11得票数 0
回答已采纳
我正在尝试对我的kafka消费者进行单元测试。我正在尝试使用kafka-client java api附带的MockConsumer类。下面是我的配置代码
@Bean
public MockConsumer consumer(){
MockConsumer consumer = new MockConsumer(OffsetResetStrategy.LATEST);
consumer.assign(Arrays.asList(new TopicPartition("test-topic", 0)));
HashMap<TopicPartiti
浏览 0提问于2017-06-20得票数 3
1回答
有一个基本的例子，它对1个消费者来说就像一个护身符。它接收消息。但是添加一个额外的消费者将被忽略。
let kafka = require('kafka-node');
let client = new kafka.Client();
let producer = new kafka.Producer(client);
let consumer1 = new kafka.Consumer(client,[ {topic: 'topic1', partition: 0}]);
let consumer2 = new kafka.Consumer(clien
浏览 4提问于2017-02-02得票数 2
回答已采纳
我正在尝试从最早的偏移量中读取卡夫卡主题，然后通过python脚本对某些记录进行墓碑。由于消息数量巨大(百万+)，我希望利用多处理来使脚本在使用消息时更快。下面是脚本中的一段代码：
from kafka import KafkaConsumer
def cleanup_kafka_topic(self, env):
# Declarations
consumer = KafkaConsumer(<topic_name>, group_id=<some_group>),
boots
浏览 71提问于2021-02-19得票数 0
这是两个部分的问题。
一、守则是：
import java.time.Duration
import java.util
import java.util.Properties
import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import scala.collection.Ja
浏览 6提问于2020-03-05得票数 1
回答已采纳
我提交了多个spark streaming作业，这些作业使用相同的group.id消耗同一个kafka topic，正如纯Kafka文档中所说的，拥有相同group.id的多个消费者将加入同一个消费组，并且kafka topic中分块的记录将被拆分成这些消费者。然而，正如我在我的工作中测试的那样，这两个spark streaming作业仍然使用相同的主题的所有分区(不是均匀分割的)，并且在who进程期间没有发生重新分区/异常。这里有人知道Spark是如何管理与纯kafka平台不同的kafka分区偏移的吗？也许这是因为zookeeper管理spark中的kafka偏移量，而不是kafka本身管
浏览 63提问于2021-07-30得票数 0
我使用的消费者组只有一个消费者，只有一个经纪人( docker wurstmeister image )。在代码中决定是否提交偏移量-如果代码返回错误，则不提交消息。我需要确保系统不会丢失任何消息-即使这意味着永远重试相同的消息(现在;)。为了测试这一点，我创建了简单的处理程序，它不会在'error‘字符串作为消息发送到kafka的情况下提交偏移量。所有其他字符串都被提交。
kafka-console-producer --broker-list localhost:9092 --topic test
>this will be commited
现在正在运行
kafka-run
浏览 1提问于2018-03-19得票数 2
2回答
嗨，我是卡夫卡CLI的工作，以了解卡夫卡如何工作。我对消费者群体感到困惑。我已经创建了三个分区的主题。我将创建生产者来为主题提供一些数据。我第一次添加一些数据如下所示。
kafka-console-producer --broker-list 127.0.0.1:9092 --topic users
>user1
kafka-console-producer --broker-list 127.0.0.1:9092 --topic users
>user2
kafka-console-producer --broker-list 127.0.0.1:9092 --topic u
浏览 7提问于2019-11-15得票数 2
回答已采纳
2回答
我使用的是camel-kafka版本的2.14.3。下面是kafka URI： <from uri="kafka:{brokerlist}?topic={topic-name}&zookeeperHost={zookeeperHost}&zookeeperPort={zookeeperPort}&groupId={groupId-name}&consumerStreams=2" /> 请注意，我在URI选项中使用了consumerStream=2。但是，当我一次将多条消息发布到主题topic-name
浏览 85提问于2020-04-21得票数 2
回答已采纳
1回答
我正在为移动客户端实现一个后端，它提供聊天服务，包括向用户提供多个聊天室。
我决定使用Kafka来进行聊天、消息订阅和发布，因为我不能使用STOMP (我选择的是Servlet而不是Servlet)。
根据我的计划，聊天消息的交换是通过以下过程进行的。(假设Clients A and B在同一个聊天室。)
client A通过WebSocket.client B连接到chat server 1，通过WebSocket.chat server1 and 2连接到chat server 2，订阅Kafka topic 'chat'client A将聊天消息发送到chat se
浏览 5提问于2020-10-07得票数 1
1回答
这里有一个奇怪的问题，卡夫卡在一个带有一个动物园管理员实例的2个代理设置中，都在AWS上。每个代理位于单独的可用性区域(AZ)。孤独的动物园管理员实例与其中一个经纪人共享AZ。我试图模拟一个滚动重新启动的情况，我们希望数据流到Kafka，即使在经纪人倒闭的时候。
当另一个AZ经纪人被拿下的时候，我能看到一个平稳的过渡到与Zookeeper共享AZ的经纪人。但是，当我在与Zoo门将共享的AZ上取下Kafka broker时，其他AZ代理将不会使用数据。不过，我认为领导人的分配会按预期的方式进行。我不确定这是否是我们的设置，卡夫卡，或动物园管理员的问题。
我们为代理间通信启用了SSL，而且生产者也
浏览 1提问于2016-06-15得票数 1
这就是场景:我知道，使用与Spring相关的最新API (比如kafka 2.10)，我们可以这样做：
@KafkaListener(id = "id0", topicPartitions = { @TopicPartition(topic = "SpringKafkaTopic", partitions = { "0" }) })
@KafkaListener(id = "id1", topicPartitions = { @TopicPartition(topic = "SpringKafkaTopic",
浏览 0提问于2017-04-17得票数 1
例如，假设我有一个有4个分区的主题。我向这个话题发送了4k条信息。每个分区获得1k条消息。由于外部因素，消费者中有3人分别处理所有1k条信息。然而，第四个分区只能通过200条消息，留下800条消息留给处理。是否有一种机制可以让我“重新平衡”主题中的数据，比如给分区1-3200个分区4s数据，让所有分区都有200个消息作为一个过程？
我并不是在寻找向使用者组添加额外节点的方法，而是让kafka平衡分区。
从重新分配分区添加的输出：
当前分区副本分配
{
"version": 1,
"partitions": [
{
"topic
浏览 0提问于2018-04-23得票数 3
回答已采纳
2回答
这是我的kafka消息制作人：
ProducerRecord producerRecord = new ProducerRecord(topic, "k1", message);
producer.send(producerRecord);
这是我的客户
TopicPartition partition0 = new TopicPartition(topic, 0);
consumer.assign(Arrays.asList(partition0));
final int minBatchSize = 200;
List&
浏览 0提问于2016-07-29得票数 0
我执行shell脚本kafka-topics.sh将分区数从3改为10。
./bin/kafka-topics.sh --zookeeper xxx.xx.xxx.xx:2181/my-broker-0 --alter --topic target-topic-example-1 --partitions 10
我一执行这个命令，Kafka集群就出现了错误。
[2018-11-20 14:45:29,954] INFO Created log for partition [target-topic-example-1,5] in /data/kafka/brokers with propert
浏览 0提问于2018-11-20得票数 0
我想为我的pyspark应用程序使用kafka自定义分割器，从kafka推送到另一个kafka主题。使用pyspark处理将数据从源转换到宿。我希望能够根据data/message中的某个键控制应该将数据推送到哪个分区。在中，我找不到此类用例的任何参考资料或示例。我正在使用python处理和pyspark，被用作kafka客户端，但它也缺乏自定义分区程序的文档/示例。
是否有解决方案可以实现这一点？
下面的spark代码尝试使用分区列，而不是按分区列推送数据。
df = spark.range(5)
df = (df
.withColumn("topic", F.l
浏览 1提问于2021-11-01得票数 1
1回答
如果我有一个主题和多个具有相同groupId的消费者，那么只有一个消费者应该收到消息(领导者)。我想选择哪个消费者将成为领导者，并在java中接收消息，但我不知道该怎么做。
这是我的一个消费者的代码。是否有一些属性可以选择领导者？
@KafkaListener(topics = {"topic_kafka","topic_kafkaaa"}, groupId = "group_id")
public void consume(String message) throws IOException {
log.info(
浏览 1提问于2019-08-22得票数 0
卡夫卡错误的含义是什么？
2018-08-22 11:40:49,429警告消费者客户端=消费者-1，群40=控制台-消费者-62114一个分区有没有匹配的侦听器的代理，包括
我跑步的时候就明白了：
./kafka-console-consumer.sh --topic topicname --bootstrap-server localhost:9094
当我试图阅读这个主题时，我在一个golang程序中遇到了一些错误：
2018/08/22 11:44:12 ReadOffsetWithRetryOnError conn错误：<拨号tcp :0:连接:连接拒绝> ka
浏览 0提问于2018-08-22得票数 18
回答已采纳
我正在尝试使用Cassandra接收器连接器读取2个kafka主题，并插入到2个Cassandra表中。我该怎么做呢？
这是我的connector.properties文件：
name=cassandra-sink-orders
connector.class=com.datamountaineer.streamreactor.connect.cassandra.sink.CassandraSinkConnector
tasks.max=1
topics=topic1,topic2
connect.cassandra.kcql=INSERT INTO ks.table1 SELECT * FR
浏览 6提问于2018-02-23得票数 6
回答已采纳
2回答
我想使用Pyhon和Kafka客户端获取Kafka主题的承诺偏移量和结束偏移量(位置)，
我想知道为什么偏移量由topic + consumer-group本身(KafkaAdminClient.list_consumer_group_offsets(consumer_group))接收
而位置是由消费者+主题的分区(KafkaConsumer.end_offsets(partitions))接收的。
有谁有线索吗？
浏览 3提问于2020-06-07得票数 0
嗨，我是Spring Boot @kafkaListener的新手。服务A不断在kafka topic上发布消息。我的服务使用来自那个topic的消息。两个服务(服务A和我的服务)中的topic分区是相同的，但是与发布消息相比，使用消息的比率很低。我可以在lag中看到kafka。
我怎样才能填补这个空白呢？或者如何提高信息的消耗率？
可以有单独的线程来处理消息。我可以在队列中使用一条消息(添加到队列后确认)，另一个线程将从该队列中读取以处理该消息。
是否有任何由Spring提供的设置或属性来提高consumption?的速率？
浏览 2提问于2021-10-19得票数 0
2回答
kafka增加了在连接器中使用regex的新功能，但是似乎连接器启动后新添加的主题数据直到连接器重新启动时才会被消耗。我们需要动态添加新的主题，并让连接器根据连接器属性中定义的正则表达式使用该主题。怎样才能实现呢？例如: regex: topic-.* topic: topic-1，topic-2如果我引入了新的topic topic-3，那么如何让连接器在不重新启动的情况下使用主题数据？
浏览 11提问于2018-07-31得票数 3
我试图设置使用者组属性，使其只能每分钟使用一条消息。
consumer.properties
group.id=test_group
max.poll.interval.ms=60000
session.timeout.ms=60000
max.poll.records=1
生产者
kafka-console-producer.sh --broker-list localhost:6667 --security-protocol SASL_PLAINTEXT --topic cdc_poc
Consumer
kafka-console-consumer.sh --bootstrap-serv
浏览 2提问于2019-03-01得票数 1
回答已采纳
1回答
我需要一个建议，什么是正确的卡夫卡结构使用我的项目和原因。
我的项目是为投资机器人管理创造一个平台。非常高水平-你可以编码几个投资策略，并将它们上传到平台上，它们将实时执行，提供分析和性能的实时信息。这些策略是从4个数据流中输入信息的。当他们阅读四个不同的卡夫卡主题时，这些数据被传递给策略。这个kafka主题直接从交换websocket接收信息。在任何给定的时间，平台中都有一个动态的机器人数量。
我所做的是:使用图像卡夫卡-wurmeister和动物园，初始化卡夫卡初始化所有的卡卡主题，我将需要事先。我将所需的数据推送给Kafka，方法是将所有信息生成到以下主题：
payloads = [
浏览 3提问于2020-05-10得票数 0
回答已采纳
我试图使用相同的@KafkaListener实现来消费来自Kafka的多个主题，但我希望每个主题都有一个消费者(每个消费者只有一个分区)。为了实现这一点，我创建了一个包含topicPattern="topic1|topic2|topic3"的@KafkaListener、一个包含并发3的ConcurrentKafkaListenerContainerFactory和一个groupId consumer_group。问题是所有的主题都被分配给同一个消费者，而另外两个消费者变得空闲，如下日志所示：
21:49:13.140 [org.springframework.kafka.Ka
浏览 2提问于2017-07-28得票数 1
1回答
我正在Linux上测试Kafka，但我不知道出了什么问题，因为测试结果与我所理解的不同。
让我来解释一下背景。
目前，三个代理程序分别使用9092、9093和9094端口在centos7中配置了KafkaVersion2.8.1。对于生产者，所有三个端口都连接到引导服务器设置，然后执行。
kafka-console-producer.bat --broker-list localhost:9092,localhost:9093,localhost:9094 --topic test
就消费者而言，设立了三个，以便将其连接到这三个港口中的每一个。
1. kafka-console-consume
浏览 2提问于2022-01-05得票数 2
回答已采纳
这不是一个重复的问题，因为即使类似的问题在堆栈溢出上可用，但在我的情况下没有答案。因此，我在这里提出了我的疑问。 我想获取可用的Kafka主题列表，并检查特定主题是否存在。我想在不使用消费者和任何shell命令的情况下完成此操作。我正在通过库本身寻找解决方案。所以，为了获取我关注的kafka主题的列表- import kafka
client = kafka.KafkaClient(bootstrap_servers='localhost:9092')
topicList = client.topic_partitions 但是，在这里我得到了错误，因为- 'Kafk
浏览 66提问于2020-12-25得票数 0
我使用的是Kafka 0.9.0.1。
我第一次启动应用程序时，需要20-30秒才能检索到主题中的“最新”消息
我使用过不同的Kafka代理(具有不同的配置)，但我仍然看到这种行为。对于后续的消息，通常不会出现速度缓慢的情况。
这是预期的行为吗？通过运行此示例应用程序并将代理/主题名称更改为您自己的设置，您可以在下面清楚地看到这一点
public class KafkaProducerConsumerTest {
public static final String KAFKA_BROKERS = "...";
public static final Stri
浏览 0提问于2016-07-23得票数 2
我有一个主题叫做“测试主题”，有3个分区。
当我以组id设置为"test-group“启动使用者(使用者-1)时，它连接并从主题上的所有分区读取。到目前一切尚好。
当我在同一组中启动另一个消费者(消费者-2)时，问题就出现了。我希望这是两个消费者之间重新平衡的分割分区，例如，1得到分区0和2，消费者2得到分区1。这不会发生，我肯定会得到这些重新平衡的回调，但最后，两个消费者从同一个分区读取。
更奇怪的是，如果我启动了Kafka (kafka-console- all . as )的另一个消费者，那么它将按照预期在所有消费者之间重新平衡和划分所有分区。
我使用kafka节点并提供以下选项
浏览 0提问于2019-09-07得票数 1
我已经创建了一个节点，用于将消息发布到主题流，该主题流将由使用kafka- HighLevelProducer的ConsumerGroupStream使用。当我从相同的ConsumerGroup创建多个消费者来消费同一主题时，只创建了一个分区，并且只有一个消费者在消费。我还尝试为该主题定义分区的数量，尽管我不确定是否需要在创建主题时定义它，如果需要，我需要预先定义多少个分区。此外，是否有可能将对象推送到转换流而不是字符串(我目前使用JSON.stringify，因为否则我会在消费者中获取对象对象。
const myProducerStream = ({ kafkaHost, highWater
浏览 1提问于2020-01-07得票数 3
当我向我的使用者查询分配的主题分区列表时，结果中的所有分区的偏移量都是-1001。如果我打印出收到消息的偏移量，则偏移量被设置为正确的值。 这是我用来消费消息的代码： static void print_partition_list(FILE* fp,
const rd_kafka_topic_partition_list_t
* partitions) {
int i;
for (i = 0; i < partitions->cnt; i++) {
fprintf(fp, "%s %s [%d] offset %lld&#
浏览 220提问于2020-01-23得票数 1
回答已采纳
在我的场景中，有一个队列，其中许多资源将消息放入该队列，而许多消费者读取消息并执行特定的工作。
对于这个场景，我使用以下命令在Kafka中创建了一个主题
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test1
现在，我开发了一个java类来使用它。
public class ConsumerGroupExample {
private final ConsumerConnector consumer;
private
浏览 2提问于2015-04-11得票数 0
我以前从来没有用过kafka。我有两个访问本地kafka实例的测试Go程序:一个阅读器和一个写入器。我正在尝试调整我的生产者、消费者和kafka服务器设置，以获得特定的行为。 我的作者： package main
import (
"fmt"
"math/rand"
"strconv"
"time"
"github.com/confluentinc/confluent-kafka-go/kafka"
)
func main() {
rand.Seed(time
浏览 9提问于2019-01-24得票数 2
我在我的spring boot应用程序中创建了一个kafka消费者，它侦听一个主题- my_topic，在读取事件my_event时，我执行了一些操作。现在我是这样做的：
@KafkaListener(topics = "my_topic",
containerFactory = "my_kafka_container_factory")
public void handleMyKafkaEvents(String eventJson) {
MyDTO my_dto = gson.fromJson(eventJson, MyDTO.c
浏览 26提问于2019-06-05得票数 2
问题所在
嗨，我试着用Grafana来想象卡夫卡的滞后。我一直在尝试记录卡夫卡的滞后使用米特里贝特，并自己做的数学，因为米特里拍不支持日志卡夫卡滞后的版本，我正在使用()。我不是使用max(partition.offset.newest) - max(consumergroup.offset)来计算延迟，而是在特定的kafka.topic.name上使用过滤过的sum(partition.offset.newest) - sum(consumergroup.offset)。但是，这笔钱不算，经过进一步的调查，我发现计数甚至不算！分区偏移量为30 /10，而使用者组偏移计数为12 /10。我希望
浏览 1提问于2019-12-10得票数 0
我想用NodeJS和Kafka构建一个API，它可以将一个偏移量和一个主题作为输入，并输出从偏移量开始的前10条消息。我在和中尝试了这种方法。
它们提供的使用者API允许使用来自特定偏移量的消息。我想停止消费的信息，一旦我已经阅读了大约10条消息。但是两个API调用都将继续获取消息，直到最后一条消息。我怎么能停止那样做呢？
这是我的编辑的完整代码
var Kafka = require('no-kafka');
var express = require("express");
var app = express();
var producer = new K
浏览 7提问于2016-04-28得票数 1
我有一个spring boot应用程序，我想将一个Kafka主题的负载分布到3个open-shift pod中。我有下面的例子，我可以在三个不同的线程上监听3个Kafka分区，这个spring boot应用程序将加载到一个openshift pod中。但我希望能够在一个pod上从一个Kafka分区监听，这样当我在open-shift上加载3个pod时，每个pod将从一个Kafka分区监听。这将允许我将应用程序扩展到N个pod上的N个分区。我不确定这是否可能，或者是否需要使用不同的方法。谢谢
public class DepAcctInqConsumerController {
p
浏览 0提问于2021-03-11得票数 1
当file节拍输出数据到kafka时，文件日志中有许多警告消息。
。。
*WARN producer/broker/0 maximum request accumulated, waiting for space
*WARN producer/broker/0 maximum request accumulated, waiting for space
。。
在我的文件节拍配置中没有什么特别之处：
。。
output.kafka:
hosts: ["localhost:9092"]
topic: "log-oneday"
。。
我还在kafka中更
浏览 2提问于2017-01-23得票数 1
我目前正在Dataproc上运行星星之火作业，并且在尝试重新加入一个组并从kafka主题读取数据时出错。我已经做了一些调查，不知道问题是什么。我已经将auto.offset.reset设置为earliest，因此应该读取最早可用的未提交偏移量，最初我的星火日志如下所示：
19/04/29 16:30:30 INFO
org.apache.kafka.clients.consumer.internals.Fetcher: [Consumer
clientId=consumer-1, groupId=demo-group] Resetting offset for
partition
浏览 0提问于2019-04-29得票数 5
4回答
假设我有一个Kafka制作人。
我的消费者首先将消息M1写入Kafka上的主题T1 (通过同步api)，然后将消息M2写入主题T2。
此时，消费者正在收听主题T1和T2。每当它看到topic T2上的消息时，它都会查找topic T1上的相关消息。
是否可以保证每当消费者看到主题M2上的消息T2时，都会看到主题T1上的消息M1？(因为在M2之前编写了M1 )
浏览 2提问于2017-01-16得票数 1
1回答
因此，我有一个消费者组在运行并侦听来自某个主题的消息。如何将更多的消费者实例添加到消费组中？如果这就是我收听消息的方式 let consumer = new kafka.ConsumerGroup(options, topicname);
consumer.on("message", function(message) {
//process message here
}); 我尝试创建消费者并在选项中包含相同的组名， new kafka.Consumer(
client,
[
{ topic: topicName}
浏览 33提问于2020-10-07得票数 1
我想使用这个库来阅读主题__consumer_offsets：
我的问题是，除非我指定一个分区，否则似乎什么都不会发生。默认情况下，这个主题有100个分区，查询kafka以获取分区列表，然后循环它们来读取它们似乎是不合理的，我希望在库中有一个预先存在的方法，从主题中的所有分区读取消息。
目前，在我与kafkacat确认__consumer_offsets topic的partition 15中有消息后，以下工作正常：
r := kafka.NewReader(kafka.ReaderConfig{
Brokers: []string{"kafka:9092"},
浏览 164提问于2019-06-11得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
