java - 线程“主要” org.apache.kafka.streams.errors.InvalidStateStoreException中的异常:

我正在尝试访问在同一java程序中创建的inMemoryStore。但是返回一个异常
““线程“主”中的异常org.apache.kafka.streams.errors.InvalidStateStoreException：状态存储storeName可能已迁移到另一个实例。

当我使用persistentKeyValueStore时，它工作正常并且能够创建存储并返回值。

package com.bakdata.streams_store.demo;

import java.util.Collection;
import java.util.Properties;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.stre7ams.state.Stores;
import org.apache.kafka.streams.state.StreamsMetadata;

public class InMemoryStore {

public static void main(String[] args) throws Exception {
    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "stream-id-0001");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

    String storeName = "sample";
    KeyValueBytesStoreSupplier stateStore = Stores.inMemoryKeyValueStore(storeName);
    StoreBuilder<KeyValueStore<String, String>> storeBuilder = Stores.keyValueStoreBuilder(stateStore, Serdes.String(), Serdes.String());

    StreamsBuilder builder = new StreamsBuilder();
    builder.addStateStore(storeBuilder);
    KStream<String, String> inputStream = builder.stream("material_test1");
    KafkaStreams streams = new KafkaStreams(builder.build(), props);

    try {
        streams.start();
        Thread.sleep(30000);
    } catch (final Throwable e) {
        System.exit(1);
    }
    final ReadOnlyKeyValueStore<String, String> keyValueStore = streams.store(storeName, QueryableStoreTypes.keyValueStore());
    KeyValueIterator<String, String> range = keyValueStore.all();
    while (range.hasNext()) {
        KeyValue<String, String> next = range.next();
        System.out.println("Key: " + next.key + ", value: " + next.value);
    }
}
}

线程“主”中的异常
org.apache.kafka.streams.errors.InvalidStateStoreException：状态
存储，样本可能已迁移到另一个实例。在
org.apache.kafka.streams.state.internals.QueryableStoreProvider.getStore（QueryableStoreProvider.java:62）
在
org.apache.kafka.streams.KafkaStreams.store（KafkaStreams.java:1067）
在
com.bakdata.streams_store.demo.InMemoryStore.main（InMemoryStore.java:59）


我期望从ReadOnlyStoreQuery打印值。

最佳答案

流上不能有StateStore，因为单个键可能有多个值。您需要先将其转换为KTable（streams.table(...)）或GlobalKtable（streams.globalTable(...)）。

Kotlin示例：

val businessObjects = streamsBuilder.globalTable("topic", eventStore("store-name"))

其中eventStore是：

fun eventStore(name: String) = Materialized.`as`<String, String>(Stores.inMemoryKeyValueStore(name))
    .withKeySerde(Serdes.String())
    .withValueSerde(Serdes.String())

启动流后：

var store: ReadOnlyKeyValueStore<String, String> = streams.store("store-name", keyValueStore<String, String>())

注意：还有一个接口KafkaStreams.StateListener用于流准备就绪

 override fun onChange(newState: KafkaStreams.State?, oldState: KafkaStreams.State?) =
    Option.fromNullable(newState)
        .filter { REBALANCING == oldState && RUNNING == it }
        .map { store = streams.store("store-name", keyValueStore<String, String>()) }
        .getOrElse { log.info("Waiting for Kafka being in REBALANCING -> RUNNING, but it is $oldState -> $newState") }

或者，您也可以将流转换为KTable

stream.groupByKey().reduce(...)

如here所述。

关于java - 线程“主要” org.apache.kafka.streams.errors.InvalidStateStoreException中的异常:，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/57007888/