From 02567b1dae8bb2333506bac11f005b8e0c597cb3 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 21 Jul 2023 18:23:54 -0400 Subject: [PATCH] Kafka Streams binder error handling - Initial support for Kafka Streams binder runtime error handling - This feature is very early in it's implementation and experimental right now --- .../kafka/streams/DltAwareProcessor.java | 76 +++++++++++++++++++ .../kafka/streams/DltSenderContext.java | 48 ++++++++++++ 2 files changed, 124 insertions(+) create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltAwareProcessor.java create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltSenderContext.java diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltAwareProcessor.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltAwareProcessor.java new file mode 100644 index 000000000..d5704aa47 --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltAwareProcessor.java @@ -0,0 +1,76 @@ +/* + * Copyright 2023-2023 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka.streams; + +import java.util.function.BiFunction; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; + +import org.springframework.cloud.stream.function.StreamBridge; + +/** + * + * @author Soby Chacko + * @since 4.1.0 + */ +public class DltAwareProcessor implements Processor { + + private final BiFunction> delegateFunction; + + private ProcessorContext context; + + private final String dltDestination; + + private final DltSenderContext dltSenderContext; + + public DltAwareProcessor(BiFunction> businessLogic, String dltDestination, DltSenderContext dltSenderContext) { + this.delegateFunction = businessLogic; + this.dltDestination = dltDestination; + this.dltSenderContext = dltSenderContext; + } + + @Override + public void init(ProcessorContext context) { + Processor.super.init(context); + this.context = context; + } + + @Override + public void process(Record record) { + try { + KeyValue keyValue = this.delegateFunction.apply(record.key(), record.value()); + //TODO: What should be the timestamp? + Record downstreamRecord = new Record<>(keyValue.key, keyValue.value, System.currentTimeMillis(), record.headers()); + this.context.forward(downstreamRecord); + } + catch (Exception exception) { + StreamBridge streamBridge = this.dltSenderContext.getStreamBridge(); + if (streamBridge != null) { + streamBridge.send(dltDestination, record.value()); + } + } + } + + @Override + public void close() { + Processor.super.close(); + } + +} diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltSenderContext.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltSenderContext.java new file mode 100644 index 000000000..aa7c0f8cc --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltSenderContext.java @@ -0,0 +1,48 @@ +/* + * Copyright 2023-2023 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka.streams; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.cloud.stream.function.StreamBridge; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.context.ConfigurableApplicationContext; + +/** + * @author Soby Chacko + */ +public class DltSenderContext implements ApplicationContextAware, InitializingBean { + + private ConfigurableApplicationContext applicationContext; + + private StreamBridge streamBridge; + + @Override + public void afterPropertiesSet() { + this.streamBridge = applicationContext.getBean(StreamBridge.class); + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = (ConfigurableApplicationContext) applicationContext; + } + + public StreamBridge getStreamBridge() { + return this.streamBridge; + } +}