ctoestreich

 <dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>kafka</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>spock</artifactId>
    <scope>test</scope>
</dependency>

ctoestreich

class TestEventConsumerSpec extends AbstractEmbeddedKafkaSpec {

    def "Test message gets routed"() {
        given:
        TestConsumer consumer = embeddedServer.getApplicationContext().getBean(TestConsumer)
        TestKafkaClient client = embeddedServer.getApplicationContext().getBean(TestKafkaClient)

        when:
        SomeAvroData event = ...//create test data here

ctoestreich

class KafkaStreamsHealthDisabledSpec extends AbstractTestContainersSpec {

    def "health check disabled"() {
        when:
        def bean = context.findBean(KafkaStreamsHealth)

        then:
        !bean.isPresent()
    }

ctoestreich

import groovy.util.logging.Slf4j
import io.micronaut.context.ApplicationContext
import io.micronaut.core.util.CollectionUtils
import io.micronaut.runtime.server.EmbeddedServer
import org.apache.kafka.clients.admin.AdminClient
import org.apache.kafka.clients.admin.NewTopic
import org.testcontainers.containers.KafkaContainer
import spock.lang.AutoCleanup
import spock.lang.Shared
import spock.lang.Specification

ctoestreich

/**
 * Generates a lot of code which fits with a class that is a representation of an immutable entity.
 * <p>
 * Equivalent to {@code @Getter @FieldDefaults(makeFinal=true, level=AccessLevel.PRIVATE) @AllArgsConstructor @ToString @EqualsAndHashCode}.
 * <p>
 * Complete documentation is found at <a href="https://projectlombok.org/features/Value">the project lombok features page for &#64;Value</a>.
 * 
 * @see lombok.Getter
 * @see lombok.experimental.FieldDefaults
 * @see lombok.AllArgsConstructor

ctoestreich

# https://projectlombok.org/features/configuration

config.stopBubbling = true
lombok.addLombokGeneratedAnnotation = true
lombok.anyConstructor.addConstructorProperties = true
lombok.copyableAnnotations += javax.inject.Named
lombok.copyableAnnotations += io.micronaut.http.client.annotation.Client
lombok.copyableAnnotations += io.micronaut.context.annotation.Parameter
lombok.copyableAnnotations += io.micronaut.context.annotation.Value
lombok.copyableAnnotations += io.micronaut.configuration.hibernate.jpa.scope.CurrentSession

ctoestreich

package com.foo.http.client;

import io.micronaut.http.MediaType;
import io.micronaut.http.annotation.Get;
import io.micronaut.http.client.annotation.Client;
import io.reactivex.Flowable;
import reactor.core.publisher.Flux;

@Client("/streaming")
public interface HeadlineClient {

ctoestreich

package com.foo.event.client;

import io.micronaut.configuration.kafka.annotation.KafkaClient;
import io.micronaut.configuration.kafka.annotation.KafkaKey;
import io.micronaut.configuration.kafka.annotation.Topic;

@KafkaClient(id = "foo-kafka-client")
public interface FooEventKafkaClient {

    void publishEvent(@Topic String topic, @KafkaKey final String key, final Foo foo);

ctoestreich

package com.company.project.event.client;

import com.company.project.event.v1.FooEvent;
import io.micronaut.configuration.kafka.annotation.KafkaClient;
import io.micronaut.configuration.kafka.annotation.KafkaKey;
import io.micronaut.configuration.kafka.annotation.Topic;
import io.micronaut.context.annotation.Requires;
import io.micronaut.retry.annotation.Retryable;

@KafkaClient(id = "foo-events-kafka-client")

ctoestreich

package com.company.kafka.client;

import com.company.avro.PersonObject;
import io.micronaut.configuration.kafka.annotation.KafkaClient;
import io.micronaut.configuration.kafka.annotation.KafkaKey;
import io.micronaut.configuration.kafka.annotation.Topic;

@KafkaClient(id = "person-object-client")
public interface PersonObjectClient {