Kafka Streams - 在字段和任务/线程模型中具有状态的转换器答案

【问题标题】：Kafka Streams - Transformers with State in Fields and Task / Threading ModelKafka Streams - 在字段和任务/线程模型中具有状态的转换器
【发布时间】：2021-02-23 19:19:19
【问题描述】：

我有一个带有状态存储的变压器，它使用标点符号对所述状态存储进行操作。标点符号的几次迭代后，操作可能已经完成，所以我想取消标点符号——但仅限于实际完成分区各自状态存储操作的任务。尚未完成的任务的标点操作应继续运行。为此，我的转换器保留了对 schedule() 返回的 Cancellable 的引用。

据我所知，每个 Task 总是有自己的隔离 Transformer 实例，每个 Task 在该实例中都有自己隔离的调度 punctuate() (?)

但是，由于这是有效的状态，但不在 stateStore 中，我不确定这有多安全。例如，在某些情况下，一个转换器实例可能会跨任务共享（因此绝对不能将任何状态保留在 StateStores 之外）？


public class CoolTransformer implements Transformer {

  private KeyValueStore stateStore;
  private Cancellable taskPunctuate; // <----- Will this lead to conflicts between tasks?


  public void init(ProcessorContext context) {
    this.store =  context.getStateStore(...);
    this.taskPunctuate = context.schedule(Duration.ofMillis(...), PunctuationType.WALL_CLOCK_TIME, this::scheduledOperation);
    
  }

  private void scheduledOperation(long l) {
    stateStore.get(...)
    // do stuff...

    if (done) {
      this.taskPunctuate.cancel(); // <----- Will this lead to conflicts between tasks?
    }
  }

  public KeyValue transform(key, value) {
    // do stuff
    stateStore.put(key, value)  
  }

  public void close() {
    taskPunctuate.cancel();
  }
}

【问题讨论】：

标签： apache-kafka apache-kafka-streams

【解决方案1】：

您也许可以查看TransformerSupplier，特别是TransformSupplier#get()，这将确保我们在应该保持独立时创建新的转换器。此外，Transformers 不应共享对象，因此请注意您的 Cancellable taskPunctuate。如果违反了其中任何一种情况，您应该会看到类似 org.apache.kafka.streams.errors.StreamsException: Current node is unknown、ConcurrentModificationException 或 InstanceAlreadyExistsException 的错误。

【讨论】：