akka-kafka

0热度

1回答

我使用python-kafka的SimpleConsumer来监听kafka代理中的主题。卡夫卡经纪人在主机名为BROKER_HOST的机器上运行。现在，从代理BROKER_HOST一主题| &话题元SimpleConsumer请求得到一个元组 (Broker metadata, Topic metadata) 经纪人的元数据来作为， {0: BrokerMetadata(nodeId=0,

1热度

2回答

处理消息后提交Kafka消费者补偿的好模式是什么？

我正在使用Akka Streams Kafka将卡夫卡消息传递给远程服务。我希望保证该服务每次只收到一条消息（至少一次，最多一次发送）。这是我想出了代码： private def startFlow[T](implicit system: ActorSystem, config: Config, subscriber: ActorRef, topicPattern: Str

0热度

1回答

如何在卡夫卡主题消耗完所有可用消息之后返回包含消息列表的未来？

我可能错过了卡夫卡消费者的角度，但我想要做的是： Consumer订阅一个主题，抓住题目中的所有信息，并返回一个未来所有这些邮件列表我已经写了尝试，并做到这一点的代码是 val sink = Sink.fold[List[KafkaMessage], KafkaMessage](List[KafkaMessage]()) { (list, kafkaMessage) => list :+ ka

1热度

2回答

Akka Kafka Producersettings：重载的方法值适用于替代方案：

当我将生产者设置放入代码中时，我一次又一次地遇到问题。当我没有它时，一切正常。下面给出了它包含所有代码的文件单个文件，我试图将一个文件写入一个kafka流。并得到这个编译错误。 package somePackage import java.nio.file.Paths import akka.Done import akka.actor.{Actor, ActorLogging, Ac

0热度

1回答

阿卡卡夫卡流监理策略不工作

我运行一个阿卡流卡夫卡应用程序，我想结合的流消费者的监督策略，例如，如果经纪人下降，流消费者停止超时后死亡，主管可以重新启动消费者。这里是我的完整代码： UserEventStream： import akka.actor.{Actor, PoisonPill, Props} import akka.kafka.{ConsumerSettings, Subscriptions} import

1热度

3回答

阿卡流TCP +阿卡流卡夫卡生产者未停止不发布消息，而不是错误-ING出

我有以下流： Source(IndexedSeq(ByteString.empty)) .via( Tcp().outgoingConnection(bsAddress, bsPort) .via(Framing.delimiter(ByteString("\n"), 256, allowTruncation = true)) .map(_.utf8String)