将数据添加到状态存储以进行状态处理和容错

如何解决将数据添加到状态存储以进行状态处理和容错

我有一个执行某些状态处理的微服务。该应用程序从输入主题构造一个KStream，进行一些有状态的处理，然后将数据写入输出主题。

我将在同一组中运行3个此应用程序。当微服务发生故障时，我需要存储3个参数，接管的微服务可以查询共享状态存储并继续崩溃服务的中断位置。

我正在考虑将这三个参数推入状态存储中，并在其他微服务接管时查询数据。从我的研究中，我看到了很多示例，当人们使用状态存储执行事件计数时，但这并不是我想要的，有人知道示例还是解决此问题的正确方法？

解决方法

所以您想做两件事：

a。服务中断必须存储参数：
如果您想以一种直接的方式做到这一点，那么您所要做的就是在与状态存储相关联的主题中写一条消息（您正在使用KTable阅读该消息）。使用Kafka Producer API或KStream（可能是kTable.toStream()）来完成它。

否则，您可以手动创建状态存储：

// take these serde as just an example
Serde<String> keySerde = Serdes.String();
Serde<String> valueSerde = Serdes.String();
KeyValueBytesStoreSupplier storeSupplier = inMemoryKeyValueStore(stateStoreName);
streamsBuilder.addStateStore(Stores.keyValueStoreBuilder(storeSupplier,keySerde,valueSerde));

然后在转换器或处理器中使用它来添加项目；您必须在转换器/处理器中声明这一点：

// depending on the serde above you might have something else then String
private KeyValueStore<String,String> stateStore;

并初始化stateStore变量：

@Override
public void init(ProcessorContext context) {
  stateStore = (KeyValueStore<String,String>) context.getStateStore(stateStoreName);
}

，然后使用stateStore变量：

@Override
public KeyValue<String,String> transform(String key,String value) {
  // using stateStore among other actions you might take here
  stateStore.put(key,processedValue);
}

b。读取接管服务中的参数：
您可以使用Kafka用户使用它，但是首先要使用Kafka Streams才能使商店可用。最简单的方法是创建一个KTable；然后，您必须获取使用KTable自动创建的可查询商店名称；那么您必须实际访问商店；然后您从存储中提取一条记录值（即按其键的参数值）。

// this example is a modified copy of KTable javadocs example
final StreamsBuilder streamsBuilder = new StreamsBuilder();

// Creating a KTable over the topic containing your parameters a store shall automatically be created.
//
// The serde for your MyParametersClassType could be 
// new org.springframework.kafka.support.serializer.JsonSerde(MyParametersClassType.class) 
// though further configurations might be necessary here - e.g. setting the trusted packages for the ObjectMapper behind JsonSerde.
//
// If the parameter-value class is a String then you could use Serdes.String() instead of a MyParametersClassType serde.
final KTable paramsTable = streamsBuilder.table("parametersTopicName",Consumed.with(Serdes.String(),<<your InstanceOfMyParametersClassType serde>>));

...
// see the example from KafkaStreams javadocs for more KafkaStreams related details
final KafkaStreams streams = ...;
streams.start()
...

// get the queryable store name that is automatically created with the KTable
final String queryableStoreName = paramsTable.queryableStoreName();
// get access to the store
ReadOnlyKeyValueStore view = streams.store(queryableStoreName,QueryableStoreTypes.timestampedKeyValueStore());
// extract a record value from the store
InstanceOfMyParametersClassType parameter = view.get(key);