spring-kafka

0熱度

1回答

我正在使用spring-cloud-stream-kafka編寫一個小型壓力測試，它會產生大量事件並將它們發送到特定的Kafka主題。該主題有6個分區，我想使用用戶ID作爲分區鍵。但是，用戶ID並不總是存在，所以我希望邏輯如下所示：如果存在用戶ID分區，否則發送到隨機分區。是否有使用 spring.cloud.stream.bindings.output.producer.partitionK

1熱度

1回答

在apache kafka中複製相關ID類似功能

我們正在實施apache kafka作爲分佈式消息中間件，該中間件將被徹底用於異步通信。我有幾個使用案例，我希望能夠發佈消息給卡夫卡經紀人，並聽取對特定消息的回覆（就像我們將在兔子mq中使用相關ID）。我讀了關於apache kafka協議，好像他們在內部使用相關ID。但是我找不到任何在任何地方使用相關ID的例子。你可以請建議，如果有辦法做到這一點與Apache卡夫卡？

0熱度

2回答

春季卡夫卡 - 所有主題到同一個消費者

我試圖從卡夫卡消費多個主題使用相同的@KafkaListener實現，但我想每個主題有一個消費者（每個只有一個分區）。爲了實現這一點，我創建了@KafkaListener，其中topicPattern="topic1|topic2|topic3"和ConcurrentKafkaListenerContainerFactory具有併發3和groupId consumer_group。問題是，所有的主

2熱度

4回答

無法從外部機器連接到kafka

我開始使用Apache Kafka，並且當我試圖從外部機器連接時遇到問題。使用下面的配置，如果應用程序和docker在同一臺機器上運行，所有工作正常。但是當我把應用程序放在機器A和docker在機器B時，應用程序無法連接。我的春天卡夫卡@Configuration這行到@Bean consumerFactory和producerFactory（想象我的機器搬運工的ip = 10.10.10.

1熱度

1回答

卡夫卡重複讀

我使用卡夫卡版本0.10.2.1和春季啓動我的項目。我具有可以通過多個消費者消耗一個話題的5個分區（具有相同組ID），這些不同的機器上運行。我面對什麼問題是：我收到重複這些卡夫卡警告日誌讀取單個消息的 Auto offset commit failed for group my-consumer-group: Commit cannot be completed since the grou

1熱度

1回答

春季卡夫卡監聽器檢測器

我在春季啓動應用程序中設置了kafka監聽器，我似乎無法使用執行程序讓監聽器在池中運行。這裏是我的卡夫卡配置： @Bean ThreadPoolTaskExecutor messageProcessorExecutor() { logger.info("Creating a message processor pool with {} threads", numThreads);

0熱度

1回答

春季集成 - 卡夫卡製片人錯誤頻道

我正在使用卡夫卡製作人發佈消息給其他卡夫卡主題，其工作相當好。下面的示例模板： <int-kafka:outbound-channel-adapter kafka-template="template" channel="inputToKafka" topic="foo"/> 請問上述聲明支持errorchannel在卡夫卡消息驅動的入站通道適配

0熱度

1回答

不同的配置兩個彈簧卡夫卡消費者在一個應用程序

我是新手卡夫卡。我正在研究Spring-Kafka POC。我們的KAFKA服務器已被Kerberized化。通過所有必需的配置，我們可以訪問Kerberized Kafka服務器。現在我們有另一個要求，即必須從非Kerberized（簡單Kafka Consumer）Kafka服務器消費主題。我們可以在單個應用程序中通過創建另一個擁有自己的偵聽器的KafkaConsumer來做到這一點嗎？

3熱度

1回答

春季卡夫卡分區

下面兩個代碼片段發佈消息的行爲有什麼不同？方法1 Message<String> message = MessageBuilder.withPayload("testmsg") .setHeader(KafkaHeaders.MESSAGE_KEY, "key").setHeader(KafkaHeaders.TOPIC, "test").build(); ListenableF

1熱度

1回答

如何配置spring-kafka忽略錯誤格式的消息？

我們的卡夫卡話題有一個問題，其中DefaultKafkaConsumerFactory & ConcurrentMessageListenerContainer組合描述here與工廠使用的JsonDeserializer消耗。不幸的是，有人有點熱心，並發表了一些無效的消息到主題上。看來spring-kafka默默無法處理這些消息中的第一個。是否有可能讓spring-kafka登錄錯誤並繼續？看看記