Using Apache Kafka® and Spring Platform to Build Event-Driven Microservices

<project>
    <dependencies>
        <!-- other dependencies -->
        <dependency>
            <groupId>io.confluent</groupId>
            <artifactId>kafka-schema-registry-client</artifactId>   (1)
            <version>5.3.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro</artifactId>   (2)
            <version>1.8.2</version>
        </dependency>
        <dependency>
            <groupId>io.confluent</groupId>
            <artifactId>kafka-avro-serializer</artifactId>  (3)
            <version>5.2.1</version>
        </dependency>
        <dependency>
            <groupId>io.confluent</groupId>
            <artifactId>kafka-streams-avro-serde</artifactId>
            <version>5.3.0</version>
            <exclusions>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
    </dependencies>
    <repositories>
        <!-- other maven repositories the project -->
        <repository>
            <id>confluent</id>
            <url>https://packages.confluent.io/maven/</url> (4)
        </repository>
    </repositories>
    <plugins>
        <!-- other maven plugins in the project -->
        <plugin>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro-maven-plugin</artifactId>
            <version>1.8.2</version>
            <executions>
                <execution>
                    <phase>generate-sources</phase>
                    <goals>
                        <goal>schema</goal>
                    </goals>
                    <configuration>
                        <sourceDirectory>src/main/resources/avro</sourceDirectory> (5)
                        <outputDirectory>${project.build.directory}/generated-sources</outputDirectory>
                        <stringType>String</stringType>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</project>

@EnableBinding(Bindings.class)  (1)
@SpringBootApplication
public class KafkaStreamsApp {

  // topic parameters injection

  public static void main(String[] args) {
    final SpringApplication application = new SpringApplication(KafkaStreamsApp.class);
    application.setWebApplicationType(WebApplicationType.NONE); (2)
    application.run(args);
  }

  @Bean
  NewTopic filteredTopic() {    (3)
    return new NewTopic(topicName, partitions, replicationFactor);
  }
}

interface Bindings {

  String USERS_FILTERED = "users-filtered";
  String USERS = "users";

  @Input(USERS)     (1)
  KStream<String, User> usersI();

  @Output(USERS_FILTERED)   (2)
  KStream<String, User> filteredUsers();
}

@Component  (1)
class UserProcessor {

  @StreamListener (2)
  @SendTo(USERS_FILTERED) (3)
  KStream<String, User> processUsers(@Input(USERS) KStream<String, User> inputStream) { (4)
    return inputStream
        .filter((key, user) -> user.getAge() < 40)  (5)
        .mapValues(user -> new User(user.getName().toUpperCase(), user.getAge())) (6)
        .peek((key, user) -> log.info("New entry in filtered stream => Key = " + key + " Value = " + user)); (7)
  }
}