我遇到了 Spring Kafka 的一個問題，即它無法從行程事件訪問狀態存盤，我將該特定存盤添加到拓撲/流中。
方法一：
@Component
@RequiredArgsConstructor
@EnableKafkaStreams
@Order(2)
public class TimelineVersionUpdatedStream implements EventStream {
private static final Logger logger =
LoggerFactory.getLogger(TimelineVersionUpdatedStream.class);
@Autowired
private StreamConfiguration configuration;
@Autowired
private TimeLineChangesCaptureService timeLineChangesCaptureService;
@Autowired
public void TimelineVersionUpdatedProccess(StreamsBuilder builder) {
final Serde<String> stringSerde = Serdes.String();
final SpecificAvroSerde<TimelineVersionUpdated> timelineVersionUpdatedSpecificAvroSerde = new SpecificAvroSerde<>();
timelineVersionUpdatedSpecificAvroSerde.configure(getSerdeConfig(), false);
final SpecificAvroSerde<PaymentChanged> paymentChangedSpecificAvroSerde = new SpecificAvroSerde<>();
paymentChangedSpecificAvroSerde.configure(getSerdeConfig(), false);
KeyValueStoreBuilder paymentStoreBuilder = new KeyValueStoreBuilder(
Stores.persistentKeyValueStore("demo-store-2"),
stringSerde,
paymentChangedSpecificAvroSerde,
new SystemTime());
KStream<String, TimelineVersionUpdated> stream = builder.stream(
Topics.MOS_BUDGET_TIMELINE_VERSION,
Consumed.with(
stringSerde,
timelineVersionUpdatedSpecificAvroSerde
));
StreamsBuilder stateStore = builder.addStateStore(paymentStoreBuilder);
stream.process(new ProcessorSupplier<>() {
@Override
public Processor<String, TimelineVersionUpdated> get() {
return new Processor<>() {
private ProcessorContext context;
private KeyValueStore<String, PaymentChanged> store;
@Override
public void init(ProcessorContext processorContext) {
this.context = processorContext;
this.store = context.getStateStore("demo-store-2");
}
@Override
public void process(String s, TimelineVersionUpdated timelineVersionUpdated) {
logger.info("TimelineVersionUpdatedStream.TimelineVersionUpdatedProccess record key {} value{}", s, timelineVersionUpdated.toString());
if (timelineVersionUpdated == null) {
return;
}
timeLineChangesCaptureService.captureTimeLineChanges(timelineVersionUpdated, store);
}
@Override
public void close() {
}
};
}
});
Topology topology = builder.build();
logger.info("{}", topology.describe().toString());
}
當我運行上面的代碼時，我得到了以下例外：
org.apache.kafka.streams.errors.StreamsException: failed to initialize processor KSTREAM-PROCESSOR-0000000001
at org.apache.kafka.streams.processor.internals.ProcessorNode.init(ProcessorNode.java:127) ~[kafka-streams-2.7.2.jar:na]
at org.apache.kafka.streams.processor.internals.StreamTask.initializeTopology(StreamTask.java:879) ~[kafka-streams-2.7.2.jar:na]
at org.apache.kafka.streams.processor.internals.StreamTask.completeRestoration(StreamTask.java:234) ~[kafka-streams-2.7.2.jar:na]
at org.apache.kafka.streams.processor.internals.TaskManager.tryToCompleteRestoration(TaskManager.java:494) ~[kafka-streams-2.7.2.jar:na]
at org.apache.kafka.streams.processor.internals.StreamThread.initializeAndRestorePhase(StreamThread.java:754) ~[kafka-streams-2.7.2.jar:na]
at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:636) ~[kafka-streams-2.7.2.jar:na]
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:564) ~[kafka-streams-2.7.2.jar:na]
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:523) ~[kafka-streams-2.7.2.jar:na]
Caused by: org.apache.kafka.streams.errors.StreamsException: Processor KSTREAM-PROCESSOR-0000000001 has no access to StateStore demo-store-2 as the store is not connected to the processor. If you add stores manually via '.addStateStore()' make sure to connect the added store to the processor by providing the processor name to '.addStateStore()' or connect them via '.connectProcessorAndStateStores()'. DSL users need to provide the store name to '.process()', '.transform()', or '.transformValues()' to connect the store to the corresponding operator, or they can provide a StoreBuilder by implementing the stores() method on the Supplier itself. If you do not add stores manually, please file a bug report at https://issues.apache.org/jira/projects/KAFKA.
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.getStateStore(ProcessorContextImpl.java:149) ~[kafka-streams-2.7.2.jar:na]
at au.com.mybudget.mos.mostimelinekafkaetl.transport.stream.TimelineVersionUpdatedStream$1$1.init(TimelineVersionUpdatedStream.java:92) ~[classes/:na]
at org.apache.kafka.streams.processor.internals.ProcessorAdapter.init(ProcessorAdapter.java:57) ~[kafka-streams-2.7.2.jar:na]
at org.apache.kafka.streams.processor.internals.ProcessorNode.lambda$init$0(ProcessorNode.java:120) ~[kafka-streams-2.7.2.jar:na]
at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:883) ~[kafka-streams-2.7.2.jar:na]
at org.apache.kafka.streams.processor.internals.ProcessorNode.init(ProcessorNode.java:117) ~[kafka-streams-2.7.2.jar:na]
... 7 common frames omitted
然后我嘗試添加如下商店： 方法 2：
@Bean
public StreamsBuilderFactoryBeanCustomizer customizer() {
final Serde<String> stringSerde = Serdes.String();
final SpecificAvroSerde<PaymentChanged> paymentChangedSpecificAvroSerde = new SpecificAvroSerde<>();
paymentChangedSpecificAvroSerde.configure(Collections.singletonMap(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, getSchemaRegistryUrl()), false);
return factoryBean -> {
try {
final StreamsBuilder streamsBuilder = factoryBean.getObject();
streamsBuilder.addStateStore(Stores.keyValueStoreBuilder(
Stores.persistentKeyValueStore("store-demo-3"),
stringSerde,
paymentChangedSpecificAvroSerde
));
} catch (Exception e) {
logger.error("StreamsBuilderFactoryBeanCustomizer exception:{}", e.getMessage());
}
};
}
然后嘗試從行程中訪問該商店，但最終得到相同的例外。
請幫助理解問題。
uj5u.com熱心網友回復：
將狀態存盤添加到Topologyid 只是第一步，但它不會使其可用：為了允許 aProcessor使用狀態存盤，您必須連接兩者。
最簡單的方法是在添加時傳入狀態存盤名稱Processor：
stream.process(..., "storeName");
轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/432708.html
標籤：爪哇 弹簧靴 阿帕奇卡夫卡 阿帕奇卡夫卡流 春天卡夫卡