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

Question

我有一个带有状态存储的转换器，它使用标点符号来操作所述状态存储。重复几次标点符号之后，操作可能已经完成，所以我想取消标点符号——但只针对实际完成分区各自状态存储上操作的任务。尚未完成的任务的标点操作应继续运行。为此，我的transformer保留了对schedule（）返回的可取消项的引用。

据我所知，每个任务总是有自己的隔离Transex实例，每个任务都有自己的隔离计划标点符号（）在该实例中（？）

但是，由于这实际上是状态，但不是在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();
  }
}

Answer 1

您也许可以查看TransformerProviier，特别是TransformProviier#get（），这将确保我们在应该保持独立的时候创建新的变压器。此外，变形金刚不应共享对象，因此请注意您的Cempable taskPunctuate。如果违反了这些情况中的任何一种，您应该会看到诸如org.apache.kafka.streams.errors.StreamsException：当前节点未知、ConvoltModificationException或InstanceAlreadyExistsException之类的错误。