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; + } +}