SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DltAwareProcessor enhancements

Browse Source 
Cleaning up the code for the custom DltAwareProcessor
This commit is contained in:
Soby Chacko
2023-09-06 20:36:35 -04:00
parent 149714062c
commit d822129d67
5 changed files with 183 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/pom.xml
View File

@@ -76,6 +76,11 @@
</exclusions>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

58
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltAwareProcessor.java
View File

@@ -30,44 +30,88 @@ import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Custom {@link Processor} implementation that is capable of sending a record
* to a DLT if the processing fails.
*
* @author Soby Chacko
* @since 4.1.0
*/
public class DltAwareProcessor<KIn, VIn, KOut, VOut> implements Processor<KIn, VIn, KOut, VOut> {
/**
* Delegate {@link BiFunction} that is responsible for processing the data.
*/
private final BiFunction<KIn, VIn, KeyValue<KOut, VOut>> delegateFunction;
/**
* Event time for the forwarded downstream record.
*/
private final Supplier<Long> recordTimeSupplier;
/**
* DLT destination.
*/
private String dltDestination;
private DltSenderContext dltSenderContext;
/**
* {@link DltPublishingContext} used for DLT publishing needs.
*/
private DltPublishingContext dltPublishingContext;
/**
* A {@link BiConsumer} that does the recovery of a failed record.
*/
private BiConsumer<Record<KIn, VIn>, Exception> processorRecordRecoverer;
/**
* {@link ProcessorContext} used in the processor.
*/
private ProcessorContext<KOut, VOut> context;
/**
*
* @param delegateFunction {@link BiFunction} to process the data
* @param dltDestination DLT destination
* @param dltPublishingContext {@link DltPublishingContext}
*/
public DltAwareProcessor(BiFunction<KIn, VIn, KeyValue<KOut, VOut>> delegateFunction, String dltDestination,
DltSenderContext dltSenderContext) {
this(delegateFunction, dltDestination, dltSenderContext, System::currentTimeMillis);
DltPublishingContext dltPublishingContext) {
this(delegateFunction, dltDestination, dltPublishingContext, System::currentTimeMillis);
}
/**
*
* @param delegateFunction {@link BiFunction} to process the data
* @param dltDestination DLT destination
* @param dltPublishingContext {@link DltPublishingContext}
* @param recordTimeSupplier Supplier for downstream record timestamp
*/
public DltAwareProcessor(BiFunction<KIn, VIn, KeyValue<KOut, VOut>> delegateFunction, String dltDestination,
DltSenderContext dltSenderContext, Supplier<Long> recordTimeSupplier) {
DltPublishingContext dltPublishingContext, Supplier<Long> recordTimeSupplier) {
this.delegateFunction = delegateFunction;
this.recordTimeSupplier = recordTimeSupplier;
Assert.isTrue(StringUtils.hasText(dltDestination), "DLT Destination topic must be provided.");
this.dltDestination = dltDestination;
Assert.notNull(dltSenderContext, "DltSenderContext cannot be null");
this.dltSenderContext = dltSenderContext;
Assert.notNull(dltPublishingContext, "DltSenderContext cannot be null");
this.dltPublishingContext = dltPublishingContext;
}
/**
*
* @param delegateFunction {@link BiFunction} to process the data
* @param processorRecordRecoverer {@link BiConsumer} that recovers failed records
*/
public DltAwareProcessor(BiFunction<KIn, VIn, KeyValue<KOut, VOut>> delegateFunction,
BiConsumer<Record<KIn, VIn>, Exception> processorRecordRecoverer) {
this(delegateFunction, processorRecordRecoverer, System::currentTimeMillis);
}
/**
*
* @param delegateFunction {@link BiFunction} to process the data
* @param processorRecordRecoverer {@link BiConsumer} that recovers failed records
* @param recordTimeSupplier Supplier for downstream record timestamp
*/
public DltAwareProcessor(BiFunction<KIn, VIn, KeyValue<KOut, VOut>> delegateFunction,
BiConsumer<Record<KIn, VIn>, Exception> processorRecordRecoverer, Supplier<Long> recordTimeSupplier) {
this.delegateFunction = delegateFunction;
@@ -104,7 +148,7 @@ public class DltAwareProcessor<KIn, VIn, KOut, VOut> implements Processor<KIn, V
BiConsumer<Record<KIn, VIn>, Exception> defaultProcessorRecordRecoverer() {
return (r, e) -> {
StreamBridge streamBridge = this.dltSenderContext.getStreamBridge();
StreamBridge streamBridge = this.dltPublishingContext.getStreamBridge();
if (streamBridge != null) {
streamBridge.send(this.dltDestination, r.value());
}

7
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltSenderContext.java → binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/DltPublishingContext.java
View File

@@ -24,15 +24,18 @@ import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
/**
* The DltPublishingContext is meant to be used along with {@link DltAwareProcessor}
* when publishing failed record to a DLT. DltPublishingContext is particularly used
* for accessing framework beans such as {@link StreamBridge}.
*
* @author Soby Chacko
*/
public class DltSenderContext implements ApplicationContextAware, InitializingBean {
public class DltPublishingContext implements ApplicationContextAware, InitializingBean {
private ConfigurableApplicationContext applicationContext;
private StreamBridge streamBridge;
@Override
public void afterPropertiesSet() {
this.streamBridge = applicationContext.getBean(StreamBridge.class);

4
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java
View File

@@ -394,8 +394,8 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
@Bean
@ConditionalOnMissingBean
public DltSenderContext dltSenderContext() {
return new DltSenderContext();
public DltPublishingContext dltSenderContext() {
return new DltPublishingContext();
}
@Configuration(proxyBeanMethods = false)

120
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DltAwareProcessorTests.java Normal file
View File

@@ -0,0 +1,120 @@
/*
* 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.integration;
import java.util.Map;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.streams.kstream.KStream;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.binder.kafka.streams.DltAwareProcessor;
import org.springframework.cloud.stream.binder.kafka.streams.DltPublishingContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.condition.EmbeddedKafkaCondition;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Soby Chacko
*/
@EmbeddedKafka(topics = "hello-dlt-1")
public class DltAwareProcessorTests {
private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker();
private static Consumer<String, String> consumer;
@BeforeAll
public static void setUp() {
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "false",
embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(
consumerProps);
consumer = cf.createConsumer();
embeddedKafka.consumeFromEmbeddedTopics(consumer, "hello-dlt-1");
}
@AfterAll
public static void tearDown() {
consumer.close();
}
@Test
void testDltAwareProcessor() {
SpringApplication app = new SpringApplication(
DltAwareProcessorTests.PublishToDltOnErrorApplication.class);
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext context = app.run("--server.port=0",
"--spring.cloud.stream.bindings.errorStream-in-0.destination=error-stream-in",
"--spring.cloud.stream.kafka.streams.bindings.errorStream-in-0.consumer.application-id=test-error-stream-app-id",
"--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000",
"--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde"
+ "=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde"
+ "=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.binder.brokers="
+ embeddedKafka.getBrokersAsString())) {
receiveAndValidate("error-stream-in", "hello-dlt-1");
}
}
private void receiveAndValidate(String in, String out) {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(
senderProps);
try {
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic(in);
template.sendDefault("foobar");
ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer,
out);
assertThat(cr.value().contains("foobar")).isTrue();
}
finally {
pf.destroy();
}
}
@EnableAutoConfiguration
static class PublishToDltOnErrorApplication {
@Bean
public java.util.function.Consumer<KStream<String, String>> errorStream(DltPublishingContext dltSenderContext) {
return input -> input
.process(() -> new DltAwareProcessor<>((k, v) -> {
throw new RuntimeException("error");
}, "hello-dlt-1", dltSenderContext));
}
}
}