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)