我正在使用 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);
}
传递流:
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.