simples embedded Kafka test

gistfile1.txt

  .   ____          _            __ _ _
 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: Spring Boot ::        (v1.5.9.RELEASE)

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <parent>
        <relativePath/>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>1.5.9.RELEASE</version>
    </parent>

    <artifactId>kafka-demo</artifactId>

    <properties>
        <!-- overwrite versions from parent -->
        <!-- ... dependencies -->
        <assertj.version>3.8.0</assertj.version>

        <kafka.artifactId>kafka_2.11</kafka.artifactId>
        <kafka.version>1.0.0</kafka.version>
        <spring-integration-kafka.version>1.3.1.RELEASE</spring-integration-kafka.version>
        <spring-kafka.version>1.3.2.RELEASE</spring-kafka.version>
        <zookeeper.version>3.4.10</zookeeper.version>
        <zkclient.version>0.10</zkclient.version>
    </properties>

    <build>
        <plugins>
            <plugin>
                <artifactId>maven-enforcer-plugin</artifactId>
                <configuration>
                    <rules>
                        <dependencyConvergence/>
                    </rules>
                </configuration>
                <executions>
                    <execution>
                        <id>check</id>
                        <goals>
                            <goal>enforce</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
                <version>${kafka.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
                <version>${kafka.version}</version>
                <classifier>test</classifier>
            </dependency>

            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>${kafka.artifactId}</artifactId>
                <version>${kafka.version}</version>
                <exclusions>
                    <exclusion>
                        <groupId>org.slf4j</groupId>
                        <artifactId>slf4j-log4j12</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>${kafka.artifactId}</artifactId>
                <version>${kafka.version}</version>
                <classifier>test</classifier>
            </dependency>

            <dependency>
                <groupId>org.apache.zookeeper</groupId>
                <artifactId>zookeeper</artifactId>
                <version>${zookeeper.version}</version>
                <exclusions>
                    <exclusion>
                        <groupId>log4j</groupId>
                        <artifactId>log4j</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
            <dependency>
                <groupId>com.101tec</groupId>
                <artifactId>zkclient</artifactId>
                <version>${zkclient.version}</version>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <dependencies>
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-kafka</artifactId>
            <version>${spring-integration-kafka.version}</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.kafka</groupId>
                    <artifactId>kafka_2.10</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka-test</artifactId>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <classifier>test</classifier>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>${kafka.artifactId}</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>${kafka.artifactId}</artifactId>
            <classifier>test</classifier>
        </dependency>

        <dependency>
            <groupId>commons-logging</groupId>
            <artifactId>commons-logging</artifactId>
            <version>1.2</version>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
</project>

SimplestEmbeddedKafkaIT.java

import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.assertj.core.api.Assertions.assertThat;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutionException;

import org.junit.ClassRule;
import org.junit.Test;
import org.junit.rules.ExternalResource;
import org.junit.rules.RuleChain;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.test.context.junit4.SpringRunner;

import lombok.Getter;

@RunWith(SpringRunner.class)
@SpringBootTest(
        classes = SimplestKafkaIT.TestConfig.class,
        properties = {
                "spring.kafka.consumer.group-id=test",
                "spring.kafka.consumer.auto-offset-reset=earliest"
        })
public class SimplestKafkaIT {

    @ClassRule
    public static RuleChain chain = RuleChain
            .outerRule(new KafkaEmbedded(1, false, 1, "topic1"))
            .around(new ExternalResource() {
                @Override
                protected void before() {
                    String property = System.getProperty(KafkaEmbedded.SPRING_EMBEDDED_KAFKA_BROKERS);
                    System.setProperty("spring.kafka.bootstrapServers", property);
                }
            });

    @Autowired
    private KafkaTemplate<String, Object> template;

    @Autowired
    private MyKafkaListener listener;

    @Test
    public void testEncryptionRoundTrip() throws ExecutionException, InterruptedException {
        MILLISECONDS.sleep(2_00);

        String aMessage = "{ \"World_1" + new Random().nextLong() + "\":1}";
        template.send("topic1", aMessage)
                .get();

        MILLISECONDS.sleep(1_000);

        assertThat(listener.getMessages()).contains(aMessage);
    }

    @Configuration
    @EnableKafka
    @Import({ KafkaAutoConfiguration.class })
    public static class TestConfig {
        @Bean
        MyKafkaListener listener() {
            return new MyKafkaListener();
        }
    }

    @Getter
    public static class MyKafkaListener {
        private List<String> messages = new ArrayList<>();

        @KafkaListener(topics = "topic1")
        public void listen(String record) {
            messages.add(record);
        }
    }
}