无法在阿尔帕卡使用Transactional.Sink生成消息到Kafka主题，但我看到幂等生成器已启用

Question

您好，我正在尝试使用生产者api，如Alpakka文档中所示。我可以使用事务源使用记录，生产者被创建，但不能将消息放到主题中，不能使用阿尔帕卡中的Transactional.Sink生成主题，但我看到幂等生产者已启用。我看到日志表明它正在进入逻辑，但它没有向myTopic生成事件

clientId=producer-7fe8789c-3171-429e-afbf-d8a8ba12700c，transactionalId=7fe8789c-3171-429e-afbf-d8a8ba12700c o.a.k.c.p.KafkaProducer - Producer info启用了幂等生成器。

您能帮我解释一下为什么它不能生成主题消息吗？

我使用docker在本地运行我的代码

下面是我的代码

``` Transactional.source(consumerSettings,

        Subscriptions.topics(topicNames))

        .mapMaterializedValue(innerControl = _)

        .map(consumerRecord => {

          handleBusiness(consumerRecord)

            .flatMap(res => Source.single(res)

                   .runWith(Transactional.sink(producerSettings, 

                                       UUID.randomUUID().toString)))

        })

    }

      source.runWith(Sink.ignore)

```javascript

我的handleBusiness逻辑如下所示：

私有定义handleBusiness(consumedMessage: ConsumerMessage.TransactionalMessageString，String)：Future[EnvelopeString，String，PartitionOffset] ={

   (conversion of consumedMessage ) map { message =>

        ProducerMessage.single(new ProducerRecord("myTopic", consumedMessage.record.key, message ), consumedMessage.partitionOffset)

 }

}`

Answer 1

我可以使用一个流来做这件事，事务源也需要有一个Sink/Flow，如下所示

Transactional.source(consumerSettings,
                      Subscriptions.topics(topicNames))
                      .mapMaterializedValue(innerControl = _)
                      .mapAsync(5) { msg => business(msg)}
                      .via(Transactional.flow(producersettings, transactions-id))