在 MongoSink 响应后向 kafka 消费者提交 - alpakka mongo 连接器

Question

我正在使用 alpakka 连接器来使用来自 Kafka 的数据包并将它们插入到 Mongo db 中。在得到 Mongo db 的响应后，我试图提交偏移量，但找不到任何相同的东西。如何确保只有在数据包成功插入 Mongodb 后才会提交偏移量？

我使用 Consumer.CommittableSource 作为源，使用 MongoSink 作为接收器，并使用 RunnableGraph 运行流。请参阅代码以获得更多说明。

来源：

    public Source<ConsumerMessage.CommittableMessage<String, String>, Consumer.Control> source() {
        return Consumer.committableSource(consumerSettings, subscription);
    }

流程：

   public Flow<ConsumerMessage.CommittableMessage, String, NotUsed> transformation() {
        return Flow.of(ConsumerMessage.CommittableMessage.class).map(i -> i.record().value().toString());
    }

水槽：

    public Sink<String, CompletionStage<Done>> sink() {
        return MongoSink.insertOne(mongoCollection);
    }

图表：

        RunnableGraph graph = RunnableGraph.fromGraph(GraphDSL.create(sink(), (builder, s) -> {
            builder.from(builder.add(source()).out()).via(builder.add(transformation())).to(s);
            return ClosedShape.getInstance();
        }));
        graph.run(ActorMaterializer.create(actorSystem));

编辑：

使用 PassThroughFlow，向 Mongo 的插入工作正常，它没有给出任何异常或错误，但仍然无法提交数据包。 transformationCommit() 函数从未被调用过。

更新流程：

    public Flow<String, String, NotUsed> transformationMongo() {
        LOGGER.info("Insert into Mongo");
        return  MongoFlow.insertOne(connection.getDbConnection());
    }

public Flow<ConsumerMessage.CommittableMessage, ConsumerMessage.CommittableOffset, NotUsed> transformationCommit() {
        return Flow.of(ConsumerMessage.CommittableMessage.class).map(i -> i.committableOffset());
    }

水槽：

    public Sink<ConsumerMessage.CommittableOffset, CompletionStage<Done>> sinkCommit() {
        CommitterSettings committerSettings = CommitterSettings.create(actorSystem);
        return Committer.sink(committerSettings);
    }

PassThroughFlow：

public class PassThroughFlow {
    public Graph<FlowShape<ConsumerMessage.CommittableMessage, ConsumerMessage.CommittableMessage>, NotUsed> apply(Flow<ConsumerMessage.CommittableMessage, String, NotUsed> flow) {
        return Flow.fromGraph(GraphDSL.create(builder -> {
            UniformFanOutShape broadcast = builder.add(Broadcast.create(2));
            FanInShape2 zip = builder.add(ZipWith.create((left, right) -> Keep.right()));
            builder.from(broadcast.out(0)).via(builder.add(flow)).toInlet(zip.in0());
            builder.from(broadcast.out(1)).toInlet(zip.in1());
            return FlowShape.apply(broadcast.in(), zip.out());
        }));
    }
}

图表：

RunnableGraph graph = RunnableGraph.fromGraph(GraphDSL.create(sinkCommit(), (builder, s) -> {
            builder.from(builder.add(source()).out()).via(builder.add(passThroughFlow.apply(transformation().via(transformationMongo())))).via(builder.add(transformationCommit())).to(sales);
            return ClosedShape.getInstance();
        }));
        graph.run(ActorMaterializer.create(actorSystem));

//Insertion to mongo is working but still the packet not committed. transformationCommit() function has never been called.

Answer 1

使用MongoFlow 代替MongoSink，它可以让您在之后继续流，在那里您可以提交偏移量。