放置值时Kafka Streams状态存储异常

Question

我正在使用带有状态存储的低级处理器 API。直到 0.10.0.1 它工作正常，但我已经升级了 Kafka Streams，我得到了以下错误。我发现这是由于更改日志，它正在查看记录上下文：

java.lang.IllegalStateException: This should not happen as timestamp() should only be called while a record is processed
! at org.apache.kafka.streams.processor.internals.AbstractProcessorContext.timestamp(AbstractProcessorContext.java:150)
! at org.apache.kafka.streams.state.internals.StoreChangeLogger.logChange(StoreChangeLogger.java:60)
! at org.apache.kafka.streams.state.internals.ChangeLoggingKeyValueBytesStore.put(ChangeLoggingKeyValueBytesStore.java:47)
! at org.apache.kafka.streams.state.internals.ChangeLoggingKeyValueStore.put(ChangeLoggingKeyValueStore.java:66)
! at     org.apache.kafka.streams.state.internals.MeteredKeyValueStore$2.run(MeteredKeyValueStore.java:67)

---

@Override
    public void process(String arg0, List<Data> data {
        data.forEach((x) -> {
            String rawKey = x.getId();
            Data data = kvStore.get(rawKey);
            long bytesize = data == null ? 0 : data.getVolume();
            x.addVolume(bytesize);
            kvStore.put(rawKey, x);
        });
    }
  
public void start() {
        builder = new KStreamBuilder();
        storeSupplier =     Stores.create(getKVStoreName()).withKeys(getProcessorKeySerde()).withValues(getProcessorValueSerde()).persistent().build();
        builder.addStateStore(storeSupplier);
        stream = builder.stream(Serdes.String(), serde(),getTopicName());
        processStream(stream);
        streams = new KafkaStreams(builder, props);
        streams.cleanUp();
        streams.start();
    }

    @Override
    public void init(ProcessorContext context) {
        super.init(context);
        this.context = context;
        this.context.schedule(timeinterval);
        this.kvStore = (KeyValueStore) context.getStateStore(getKVStoreName());
    }

Answer 1

在多个流线程或分区中使用Processor 的相同实例时可能会出现此类异常。

确保您将新实例返回到ProcessorSupplier：

new ProcesorSupplier(() -> new Processor(...

这同样适用于Transformer 和TransformerSupplier。

广泛引用文档：

创建单个 Processor/Transformer 对象并在 ProcesorSupplier/TransformerSupplier#get() 中返回相同的对象引用将违反供应商模式并导致运行时异常。