为了从kafka和向kafka传输mqtt消息，我使用了一个mqtt connect worker，与此类似。不同的是，我读写mqtt代理。这很有效。

┌────────────┐        ┌────────────┐               
│            │        │            │               
│    KAFKA   │        │     MQTT   │               
│   ┌────────│        │    BROKER  │               
│   │MQTT-Con│<──────>│            │               
└───┴────────┘        └────────────┘

下一步，我想从一个kafa代理切换到一个kafka集群（现在是两个kafka代理）。

┌─────────────────────┐ 
  │    KAFKA CLUSTER    │
  │  ┌────────────┐     │                 
  │  │            │     │                 
  │  │  KAFKA A   │     │                 
  │  │   ┌────────│     │  ┌────────────┐               
  │  │   │MQTT-Con│<────│─>│            │               
  │  └───┴────────┘     │  │     MQTT   │               
  │                     │  │    BROKER  │
  │  ┌────────────┐  ┌──│─>│            │
  │  │            │  │  │  └────────────┘
  │  │  KAFKA  B  │  │  │                
  │  │   ┌────────│  │  │                 
  │  │   │MQTT-Con│<─┘  │                   
  │  └───┴────────┘     │                 
  └─────────────────────┘

我的问题是，我能做什么，我不结束与Kafkaa和Kafkab相同的信息复制？
我的mqtt消息应该是qos 2，正好是一次。