我做了两个Spring项目。一位是卡夫卡制作人，另一位是听众。然后，我有一个这样的dockerfile :在这里，我创建一个容器的动物园管理员和卡夫卡，还有一个生产者容器和两个消费者容器。
version: '3'
services:
zookeeper:
image: wurstmeister/zookeeper
container_name: zookeeper
restart: always
ports:
- 2181:2181
kafka:
image: wurstmeister/kafka
co
浏览 6提问于2022-05-11得票数 1
回答已采纳
我想删除一个Kakfa消费者组，这样当应用程序创建一个消费者并订阅一个主题时，它可以从主题数据的开头开始。 这是使用当前最新的Confluent平台3.1.2的单节点vm开发，该平台使用Kafka 0.10.1.1。 我尝试正常的语法： sudo /usr/bin/kafka-consumer-groups --new-consumer --bootstrap-server localhost:9092 --delete --group my_consumer_group 我得到了错误： Option [delete] is only valid with [zookeeper]. Note
浏览 253提问于2017-03-03得票数 20
回答已采纳
我有一个应用程序下载特定的网页内容，从一个从1 Kafka产生的URL流-生产者。我已经创建了一个有5个分区的主题，有5个kafka-consumers。然而，网页下载的超时时间是60秒。当其中一个url正在下载时，服务器会假定该消息已丢失，并将数据重新发送给不同的消费者。
我试过了我提到的所有东西
和
但我每次都收到不同的错误。
是否可以在kafka中将特定的消费者与分区绑定？我在我的应用程序中使用kafka-python
浏览 4提问于2017-08-29得票数 4
回答已采纳
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
问题所在
嗨，我试着用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
2回答
我想以编程的方式重读所有的Kafka事件。我知道有一个Application Reset Tool，但据我所知，这需要我关闭我的应用程序。我不能在我的用例中关闭应用程序。 有没有办法让我的应用程序重新读取关于Kafka主题的所有事件？示例或代码片段将非常感谢。优选地，但不一定使用Kafka流。
浏览 33提问于2020-09-01得票数 0
回答已采纳
我正在尝试手动将分区分配给消费者组中的每个消费者。但是，当您在组中添加使用者时，Kafka (或者至少是kafka-python)假定您希望组协调器完成所有任务。有没有办法仍然保留使用使用者组的所有好处(特别是自动提交)，同时能够手动管理分配分区？我能写一个自定义的分区分配器吗？
我使用的是kafka-python 1.3.3库。
浏览 8提问于2017-09-20得票数 2
4回答
AFAIK
在kafka中引入了分区和(使用者)组的概念来实现并行性。我正在通过python使用kafka。我有一个特定的主题，它有(比如说)2个分区。这意味着，如果我启动一个包含2个消费者的消费者组，他们将被映射(订阅)到不同的分区。
但是，在python中使用kafka库时，我遇到了一个奇怪的问题。我用本质上相同的group-I启动了两个消费者，并启动了让它们消费消息的线程。
但是，kafka-stream中的每一条消息都被他们两个消费了！这在我看来是荒谬的，甚至在概念上也是不正确的。有没有办法可以手动将使用者映射到某些(不同的)分区(如果它们没有自动映射到不同的分区)？
代码如下：
fr
浏览 2提问于2016-07-15得票数 1
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
回答已采纳
1回答
我可以使用KafkaConsumer在单独的线程中使用消息。
但是，当我使用multiprocessing.Process而不是threading.Thread时，我会得到一个错误：
OSError: [Errno 9] Bad file descriptor
这个和建议使用多处理并行地使用消息是可能的。有谁能给我举个例子吗？
编辑
这是一些示例代码。不好意思，原来的代码太复杂了，所以我在这里创建了一个示例，希望能够告诉您正在发生的事情。如果我使用threading.Thread而不是multiprocessing.Process，这段代码可以正常工作。
from multiprocessin
浏览 3提问于2017-09-29得票数 4
回答已采纳
我想使用这个库来阅读主题__consumer_offsets：
我的问题是，除非我指定一个分区，否则似乎什么都不会发生。默认情况下，这个主题有100个分区，查询kafka以获取分区列表，然后循环它们来读取它们似乎是不合理的，我希望在库中有一个预先存在的方法，从主题中的所有分区读取消息。
目前，在我与kafkacat确认__consumer_offsets topic的partition 15中有消息后，以下工作正常：
r := kafka.NewReader(kafka.ReaderConfig{
Brokers: []string{"kafka:9092"},
浏览 164提问于2019-06-11得票数 0
1回答
我想为三个具有相同队列的类似应用程序设置一个kafka集群类型的设置，这些队列包括AppA -> {TopicX、TopicY、TopicZ}、AppB -> {TopicX、TopicZ}、AppC -> {TopicX、TopicY}。生产者和消费者将是特定的应用程序。我使用三个具有分区1、2、3的代理在三个具有不同端口的不同配置文件中设置kafka集群。然后启动kafka服务器(集群)
我使用的是的kafka php包装器。
所以我使用了应用程序A的生产者代码
$producer->setRequireAck(-1);
$produce
浏览 4提问于2016-05-02得票数 0
回答已采纳
我使用Kafka Rest代理来生成和使用Azure APIM从Kafka主题消费，因此使用Rest代理的原因。我可以使用auto.offset.reset =最早的线性方式使用消息，所以每次调用Kafka Rest都会给我最新的消息，这是很棒的，
在以下情况下：
A1 is data requestor
B2 is API that fulfills A1's request using REST proxy for Kafka
C3 is Kafka Rest proxy server
如果A1向B2请求数据，而B2使用卡夫卡主题C3提供的消息给A1，但由于某些互联网问题，该消息
浏览 2提问于2022-07-21得票数 0
我正在运行kafka_2.11-0.9.0.0和一个基于java的生产者/消费者。消息大小约为70KB时，一切工作正常。然而，在生产者将更大的70MB消息排入队列后，kafka似乎停止向消费者传递消息。即不仅不递送大消息，而且还递送随后的较小消息。我知道生产者成功了，因为我使用了kafka回调进行确认，并且我可以在kafka消息日志中看到消息。
kafka配置自定义更改：
message.max.bytes=200000000
replica.fetch.max.bytes=200000000
消费者配置：
props.put("fetch.message.max.bytes"
浏览 1提问于2016-02-03得票数 0
1回答
我有下面的卡夫卡设置
Number of producer : 1
Number of topics : 1
Number of partitions : 2
Number of consumers : 3 (with same group id)
Number of Kafka cluster : none(single Kafka server)
Zookeeper.session.timeout : 1000
Consumer Type : High Level Consumer
生产者生成没有任何特定分区逻辑的消息(默认分区逻辑)
浏览 1提问于2014-10-28得票数 1
1回答
有时，kafka-node消费者从偏移量0开始消费，而它的默认行为是只消费较新的消息。则它将不会切换回其默认行为。你知道如何解决这个问题吗?知道发生了什么，它的行为突然改变了吗？代码非常简单，而且不需要修改代码就可以实现。
var kafka = require("kafka-node") ;
Consumer = kafka.Consumer;
client = new kafka.KafkaClient();
consumer = new Consumer(client, [{ topic: "Topic_23", partition: 0}
浏览 5提问于2020-01-31得票数 3
Kafka使用者有一个配置max.poll.records，它控制在一次调用poll()及其时返回的最大记录数。我将其设置为一个非常高的数字，这样我就可以在一次投票中获得所有消息。然而，尽管该主题有更多的消息，但该调查在一次调用中仅返回数千条消息(约6000条)。
如何进一步增加单个消费者读取的消息数？
浏览 0提问于2018-08-09得票数 31
回答已采纳
我们有一个HDP集群，有3个kafka经纪人(来自hortonworks )。
我们希望运行kafka控制台使用者，以便从带有特定偏移量的主题中获得一条消息。
/usr/hdp/current/kafka-broker/bin/kafka-console-consumer.sh --zookeeper zoo01:2181 --topic lopet.lo.pm--partition 0 --offset 34537263 --max-messages 1
但我们得到了以下信息：
我们哪里错了？
Using the ConsoleConsumer with old consumer is d
浏览 3提问于2019-12-11得票数 5
回答已采纳
我使用的是Kafka 0.10.0。在处理之前，我想知道分区中记录的大小。
在0.9.0.1版本中，我使用以下代码来查找分区的latest和earliest偏移量之间的差异。在新版本中，它在检索consumer#position方法时会卡住。
package org.apache.kafka.example.utils;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import j
浏览 0提问于2016-05-31得票数 2
下面是需要的场景。
主题-1有6个分区，现在我想创建3个消费者组cg1、cg2和cg3并像这样映射它(cg1 - 0,1；cg2 - 2,3；cg3 - 4,5)。如何使用kafka-console- can . it或kafka-consumer groups.sh创建它？
甚至Kafka文档也解释了这个场景，但是没有提到如何去做。任何帮助都很感谢！
浏览 4提问于2019-11-29得票数 4
回答已采纳
我有一个关于2个代理的3个分区的主题。(Kafka版本: 0.8.1)
使用不同的用户guid (例如: FC42B34DD7658503E040970A2C437358)作为分区键批量添加消息。(约10k条消息)
加载消息时，我有一个正在运行的使用者(consumer1)，它开始很好地处理消息。
然后，我使用相同的消费者组id启动了另一个消费者(consumer2)。
我注意到的是，consumer1停止处理消息，而consumer2开始处理所有消息。
当我停止consumer2时，consumer1接管并恢复消息处理。
我希望两个消费者都能分担负载。
有什么线索可能是问题所在吗？谢谢。
浏览 0提问于2015-03-28得票数 1
2回答
我使用的是来自的消费者。kafka版本为0.10.1.0。
下面是我的消费者的配置：
kafkaClient, err := kafka.NewConsumer(&kafka.ConfigMap{
"bootstrap.servers": broker,
"group.id": "udwg20",
"session.timeout.ms": 60000,
"go.events.
浏览 0提问于2018-10-02得票数 2
我的python脚本的输入是使用者组的名称，输出应该是使用者组下的主题列表。
我的代码目前没有返回当前偏移量为-1的主题。
有没有更好的方法，可以使用Kafka-python获取消费者组下所有主题的列表。
我实际上是在寻找替换描述一个主题在一个组cmd使用Kafka工具。
from kafka import KafkaAdminClient
topics_in_groups = {};
client = KafkaAdminClient(bootstrap_server='localhost:9092')
for group in client.list_consumer_gr
浏览 0提问于2020-05-07得票数 0
2回答
我们想知道对于给定的Kafka主题，还没有被Kafka Consumer消费和确认的消息数量。
有没有办法在Java中获得给定主题(主题有10个分区)中未消费的Kafka消息的计数?？
浏览 3提问于2019-10-24得票数 3
6回答
我是Kafka的新手，正在运行一个简单的kafka消费者/生产者示例，就像和上给出的一样。当我从终端运行consumer时，consumer正在接收消息，但我无法使用Java代码进行侦听。我也在StackoverFlow (链接：，)上搜索了类似的问题，并尝试了这些解决方案，但似乎没有什么对我有效。Kafka版本: pom使用kafka_2.10-0.10.2.1和对应的maven依赖。
面向生产者和消费者的Java代码：
public class SimpleProducer {
public static void main(String[] args) throws Interrupte
浏览 6提问于2017-06-26得票数 7
我使用的是1.5.4 spring boot，java 8和kafka 2.2.0版本。我有5个代理和4个不同的主题，每50个分区。
我有一个这样的系统：
我有两三百万张唱片。我出口这些记录的2500,2500部分上的每个卡夫卡消息。我的意思是1条kafka消息由2500条记录组成，但我使用的是相同的键，这意味着所有这些记录都会进入相同的分区。我使用紧凑的主题，也有2-3百万的记录，我使用ktable - kstream连接和收集另一条记录。(此记录具有不同的键，如Id，以便重新分区)在此连接操作之后，我返回到我的原始键进行重新分区。我的kafka streams代码监听这个主题，并执行重新分
浏览 0提问于2019-11-14得票数 0
我们有没有办法通过编程的方式在Kafka Consumer中找到滞后。我不希望外部Kafka管理器工具安装和检查仪表板。
我们可以列出所有消费者组，并检查每个组的lag。
目前我们确实有检查lag的命令，它需要Kafka所在的相对路径。
Spring-Kafka、kafka-python、Kafka Admin client或使用JMX -有没有什么方法可以让我们编码并找出延迟。
我们粗心大意，没有监控进程，消费者处于僵尸状态，延迟达到50,000，这导致了很多混乱。
只有当问题出现时，我们才会想到这些情况，因为我们正在监控脚本，但不知道它会导致僵尸进程。
非常欢迎任何想法！！
浏览 0提问于2019-08-01得票数 0
1回答
我正在设置新的Kafka集群，出于测试目的，我创建了一个包含1个分区和3个副本的主题。 现在，我通过producer并行触发消息，比方说每秒50K条消息。我在一个组中创建了一个消费者，它每秒只能获取30K条消息。 我可以更改主题级别、分区级别、消费者级别的配置。 我正在通过grafana + prometheus监视一切。 您知道哪种配置或其他配置可以帮助我使用更多的数据吗？ 提前感谢
浏览 15提问于2019-04-08得票数 3
? 如果我有一个Kafka主题，其中我只有一个partitionP0，而生产者使用不同的keysk1，k2将数据写入partitionP1。假设1-1,1-2是关键字k1的值，2-1,2-2,2-3是关键字k2的值。 我现在在主题的partitionP0中有1-1,1-2,2-1,2-2,2-3 我是否可以有两个不同的consumersC1，C2侦听主题中的partitionP0并读取与C1中不同keys...as关联的值C1仅读取k1的值，而C2仅读取K2的值？
浏览 18提问于2020-09-25得票数 0
我正在使用Spring和Kafka处理一个小批处理，该批处理从Kafka主题读取json数据，将其转换为sends对象，更改值并将其发送回Kafka主题。一切都很好，但我唯一的问题是，我的消费者总是在阅读这个话题的乞求。我需要它从最后一条未消耗的信息中读出来。我已经添加了这些属性：
ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to earliest
ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to false
ConsumerConfig.GROUP_ID_CONFIG to a random value
但这似乎不起作
浏览 2提问于2021-01-21得票数 1
回答已采纳
我有一个Ingres DB with History表，它记录数据库事件，如插入、更新和删除。我有一个生产者，这将是多线程。这个生产者将读取History表，以找到要选择的表和行，然后将该行添加到Kafka主题中。现在生产者需要确保将事件添加到Kafka主题中，其方式与History表登录的方式相同。因此，使用者读取它们的顺序与它在History表中记录的顺序相同，并在Postgrace上执行它。
我可以将这些数据生成多个生产者。示例
Producer1 has message 1 to 5
producer2 has message 6 to 10
producer3 has message
浏览 1提问于2019-01-16得票数 0
1回答
我们正在使用kafka 2.12.XXXX，并且需要观察消费者在主题的所有分区上有多大的滞后。
在./kafka-consumer-groups.sh --describe中，滞后列的求和是它的精确表示吗？
浏览 5提问于2020-04-25得票数 2
在启动应用程序时，在Kafka流上出现了一些奇怪的错误。
java.lang.IllegalArgumentException: Illegal base64 character 7b
at java.base/java.util.Base64$Decoder.decode0(Base64.java:743)
at java.base/java.util.Base64$Decoder.decode(Base64.java:535)
at java.base/java.util.Base64$Decoder.decode(Base64.java:558)
at o
浏览 1提问于2020-04-01得票数 3
回答已采纳
1回答
我有一个java客户端(使用kafka客户端版本2.4.1)，它连接到一个kafka (版本2.4.1)，其中有5个主题，每个主题5个分区。我正在为每个分区创建一个线程。我的问题是，对于每个线程，所有的消费者配置都会被记录下来： ? 首先，我想知道这是否是正常行为(因为我有25个线程，我看到的是这些日志的25倍)？也许我做错了什么。目前，我的每个使用者线程都在订阅它的主题。所以我调用了subscribe方法25次，每次调用似乎都会生成这些日志。 那我能做些什么来禁用它们吗？
浏览 6提问于2020-04-07得票数 1
回答已采纳
1回答
考虑以下代码-
@Bean
public ConsumerFactory<String, String> consumerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapAddress);
props.put(
ConsumerConfig.GROUP_ID_CONFIG,
groupId);
浏览 5提问于2018-01-14得票数 3
回答已采纳
3回答
我在一个消费者组中有两个消费者，他们分配了相同的kafka主题分区。我希望从消费者B内部获得消费者A的最后一次读取偏移量。有什么想法，如何实现这一点？
浏览 2提问于2016-08-01得票数 7
回答已采纳
2回答
我有一个主题列表(目前是10个)，它们的大小在未来可能会增加。我知道我们可以从每个主题产生多个线程(每个主题)来消耗，但是在我的例子中，如果主题的数量增加，那么消耗主题的线程的数量就会增加，这是我不想要的，因为主题不会太频繁地获取数据，所以线程将处于理想状态。
有没有办法让一个消费者消费所有主题？如果是，那么我们如何实现它呢？另外，Kafka将如何维护偏移量？请建议答案。
浏览 1提问于2016-09-19得票数 32
回答已采纳
我正在寻找一种方法来消费我的Kafka主题中具有特定偏移量范围的一些消息集(假设我的分区偏移量为200 - 300，我希望消费偏移量为250-270的消息)。
我使用下面的代码，我可以在其中指定初始偏移量，但它将消耗从250到结束的所有消息。在此之前，是否有任何方法/属性可用于设置结束偏移量以使用消息。
@KafkaListener(id = "KafkaListener",
topics = "${kafka.topic.name}",
containerFactory = "kafkaManualAckLis
浏览 3提问于2020-03-06得票数 0
我在端口8081上的kafka代理上启用了JMX。当我在jConsole中查看MBean属性时，我只看到kafka.consumer的以下内容-
kafka.consumer:type=FetchRequestAndResponseMetrics,name=FetchRequestRateAndTimeMs,clientId=ReplicaFetcherThread-2-413
kafka.consumer:type=FetchRequestAndResponseMetrics,name=FetchResponseSize,clientId=ReplicaFetcherThread-0-41
浏览 3提问于2016-06-14得票数 1
我正在介绍kafka，我想知道在消费topic中的消息时如何指定分区。
我找到了几张这样的图片：
这意味着一个消费者可以消费来自多个分区的消息，但一个分区可以被单个消费者(在消费者组中)读取。
我也读过几个面向消费者的例子，看起来是这样的：
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "consumer-tutorial");
props.put(&
浏览 0提问于2017-09-29得票数 14
回答已采纳
我正在消费数据从卡夫卡主题通过火花结构化流，该主题有3个分区。由于Spark结构化流不允许您显式地提供group.id并向使用者分配一些随机id，所以我尝试检查使用者组id是否使用了kafka命令
./kafka-consumer-groups.sh --bootstrap-server kfk01.sboxdc.com:9092,kfk02.sboxdc.com:9092,kfk03.sboxdc.com:9092 --list
output
spark-kafka-source-054e8dac-bea9-46e8-9374-8298daafcd23--1587684247-drive
浏览 2提问于2019-02-28得票数 3
回答已采纳
1回答
我正在工作的火花流应用程序使用火花结构化流，汇合开源卡夫卡集群和运行星火作业在AWS EMR。我们至少有20+卡夫卡主题，将数据生成为雅夫罗格式的个别卡夫卡主题，每个主题被划分在3到4个分区之间。我正在使用SparkReadStream阅读所有20+主题(逗号分隔的主题值)。然后过滤结果DataFrame中的每个消息行，使用正确的Avro模式应用每条消息，并将得到的Dataset[T]写入S3和Cassandra。
有几个问题我找不到答案
对于所有的主题，我可以使用一个ReadStream吗？因为我在EMR中只执行一个“火花提交作业”，它会被认为是所有主题和分区的一个星火用户吗？
sp
浏览 2提问于2017-12-18得票数 4
所以基本上我看到的是，我正在使用@KafkaListener消费消息，我故意不想承认它们，所以消费者忽略它们，并在主题中侦听新的消息。因此，如果应用程序重新启动，使用者只会从最后提交的消息开始。因此，我希望避免这种情况，我希望消费者在不重新启动的情况下考虑未提交的消息。下面是我正在使用的配置和@kafkalistener。
ReceiverConfig.java
package com.config;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serializ
浏览 3提问于2019-12-05得票数 1
1回答
有没有办法让kafka在一个消费者身上拥有多个主题？ 这是我的消费者代码，它可以很好地处理单个主题 let consumer = new Consumer(
client,
[{ topic: '1234', offset: 0, partition: 0 }],
{
autoCommit: true,
fetchMaxWaitMs: 1000,
fetchMaxBytes: 1024 * 1024,
encoding: 'u
浏览 36提问于2019-06-26得票数 1
回答已采纳
我正在尝试编写一个简单的Kafka消费者来读取windows机器上的数据。但是我的消费者不能读取任何数据。生产者产生了20多条消息，但没有一条消息被消费。
下面是我为消费者编写的代码：
Properties properties= new Properties();
properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG
浏览 0提问于2020-05-13得票数 1
当我向我的使用者查询分配的主题分区列表时，结果中的所有分区的偏移量都是-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
回答已采纳
2回答
我创建了一个NodeJS应用程序将数据插入到MongoDB集合中。此数据库插入是通过使用Kafka完成的。卡夫卡-节点是我曾经称之为卡夫卡的插件。
我可以创建主题并在生产者级别向消费者发送信息。消息和主题摘自POST请求。
每次我调用这个API时，生产者都会创建一个新消息并将其发送给消费者。在每个调用中，所有以前的消息都将返回给使用者。
我使用了fromOffset: 'earliest'和fromOffset: 'latest'选项来限制以前的消息，但不起作用。
有人能给我提个建议吗？卡夫卡节点版本
"kafka-node": "^5
浏览 28提问于2022-01-03得票数 0
2回答
我正面临着一个与卡夫卡有关的问题。
我使用当前的服务(Producer)将消息发送到Kafka主题(events)。该服务使用的是用Java语言编写的kafka_2.12 v1.0.0。
我正在尝试将它与spark-streaming的示例项目集成为Consumer服务(使用Scalav0.10.0的，用kafka_2.11编写)
消息从Producer成功发送到Kafka主题。但是，我总是收到下面的错误堆栈：
Exception in thread "main" org.apache.kafka.common.errors.InconsistentGroupProtocolE
浏览 1提问于2018-07-24得票数 8
1回答
我正在使用Alpakka kafka和scala应用程序。我的Kafka在docker内部运行，我正在尝试使用我的代码在Kafka producer上发布消息。我的代码如下
def sendMsg(xmlFile: String): Future[Done] = {
futureToFutureTry {
val producer = SendProducer(producerSettings)
producer.send(new ProducerRecord("topic_name", "Key", xmlFile)).map
浏览 4提问于2020-07-17得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
