schroedermatt

@Bean
RetryTemplate retryTemplate() {
  RetryTemplate retryTemplate = new RetryTemplate()

  FixedBackOffPolicy fixedBackOffPolicy = new FixedBackOffPolicy()
  fixedBackOffPolicy.setBackOffPeriod(1000l)
  retryTemplate.setBackOffPolicy(fixedBackOffPolicy)

  SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy()
  retryPolicy.setMaxAttempts(2)

schroedermatt

plugins {
    id 'java'
    id 'com.diffplug.gradle.spotless' version '3.26.1'
    
    // other plugins
}

repositories { .. }
dependencies { .. }

schroedermatt

@Component
public class SpringContext implements ApplicationContextAware {
  private static ApplicationContext context;

  @Override
  public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
    context = applicationContext;
  }

  public static <T> T getBean(Class<T> beanClass) {

schroedermatt

public class CustomExceptionHandler implements ProductionExceptionHandler {

  private ExceptionService exceptionService;

  public ProductionExceptionHandlerResponse handle(final ProducerRecord<byte[], byte[]> record, final Exception exception) {
    boolean shouldContinue = exceptionService.handleProductionException(record, exception);
    
    return shouldContinue ? CONTINUE : FAIL;
  }

schroedermatt

KafkaStreams streams = new KafkaStreams(topology, config);

// other configurations

streams.setUncaughtExceptionHandler((Thread thread, Throwable throwable) -> {
   // examine the throwable and do something
});

streams.start();

schroedermatt

public class ContainerKafkaTestResource implements QuarkusTestResourceLifecycleManager {

    private static final KafkaContainer kafka = new KafkaContainer();

    /**
     * @return A map of system properties that should be set for the running tests
     */
    @Override
    public Map<String, String> start() {
        kafka.start();

schroedermatt

@QuarkusTest
@QuarkusTestResource(ContainerKafkaTestResource.class)
@Stereotype
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
@DisabledIfSystemProperty(named = "container-enabled", matches = "false")
public @interface QuarkusContainerKafkaTest {}

schroedermatt

@QuarkusContainerKafkaTest
public class ExampleTest {
...
}

schroedermatt

import org.springframework.kafka.test.EmbeddedKafkaBroker;

public class EmbeddedKafkaTestResource implements QuarkusTestResourceLifecycleManager {

    public EmbeddedKafkaBroker embeddedBroker;

    /**
     * @return A map of system properties that should be set for the running test
     */
    @Override

schroedermatt

@QuarkusTest
@QuarkusTestResource(EmbeddedKafkaTestResource.class)
@Stereotype
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface QuarkusEmbeddedKafkaTest {}