Kstreamplify Framework

For framework-independent Java applications:

1. Add the dependency: Include kstreamplify-core.
2. Implement the starter: Create a class extending KafkaStreamsStarter and override topology(StreamsBuilder streamsBuilder) and dlqTopic().
3. Initialize and start: In your main method, wrap your starter in a KafkaStreamsInitializer and call .start().
4. Configure properties: Place your Kafka settings in src/main/resources/application.yml under the kafka.properties key.

Important Notes:

• To enable web services (like Topology or Interactive Queries), you must define server.port in your configuration.
• The core dependency does not include a logger; you must add one to your project manually.

<!-- Dependency -->
<dependency>
    <groupId>com.michelin</groupId>
    <artifactId>kstreamplify-core</artifactId>
    <version>${kstreamplify.version}</version>
</dependency>

<!-- Implementation and Startup -->
public class MyKafkaStreams extends KafkaStreamsStarter {
    @Override
    public void topology(StreamsBuilder streamsBuilder) {
        // Define your topology here
    }

    @Override
    public String dlqTopic() {
        return "dlq_topic";
    }
}

public class MainKstreamplify {
    public static void main(String[] args) {
        KafkaStreamsInitializer initializer = new KafkaStreamsInitializer(new MyKafkaStreams());
        initializer.start();
    }
}

<!-- Configuration in application.yml -->
kafka:
  properties:
    application.id: 'myKafkaStreams'
    bootstrap.servers: 'localhost:9092'
    schema.registry.url: 'http://localhost:8081'
server:
  port: 8080

Kstreamplify simplifies testing by providing KafkaStreamsStarterTest, which wraps the Kafka Topology Test Driver.

1. Add the test dependency: Include kstreamplify-core-test with test scope.
2. Create a test class: Extend KafkaStreamsStarterTest and override getKafkaStreamsStarter() to return your application's starter implementation.
3. Setup topics: In a @BeforeEach method, use testDriver.createInputTopic and testDriver.createOutputTopic to define your test streams. Use SerdesUtils to obtain the correct Serdes for your Avro types.
4. Override properties: If you need to change default test properties (e.g., the state directory), override getSpecificProperties().

Example test case for verifying business logic and DLQ routing:

<!-- Dependency -->
<dependency>
    <groupId>com.michelin</groupId>
    <artifactId>kstreamplify-core-test</artifactId>
    <version>${kstreamplify.version}</version>
    <scope>test</scope>
</dependency>

<!-- Test Implementation -->
public class MyKafkaStreamsTest extends KafkaStreamsStarterTest {
    private TestInputTopic<String, KafkaUser> inputTopic;
    private TestOutputTopic<String, KafkaUser> outputTopic;

    @Override
    protected KafkaStreamsStarter getKafkaStreamsStarter() {
        return new MyKafkaStreams();
    }

    @BeforeEach
    void setUp() {
        inputTopic = testDriver.createInputTopic("input_topic", new StringSerializer(),
            SerdesUtils.<KafkaUser>getValueSerdes().serializer());

        outputTopic = testDriver.createOutputTopic("output_topic", new StringDeserializer(),
            SerdesUtils.<KafkaUser>getValueSerdes().deserializer());
    }

    @Test
    void shouldUpperCase() {
        inputTopic.pipeInput("1", user);
        List<KeyValue<String, KafkaUser>> results = outputTopic.readKeyValuesToList();
        assertEquals("FIRST NAME", results.get(0).value.getFirstName());
    }

    @Test
    void shouldFailAndRouteToDlqTopic() {
        inputTopic.pipeInput("1", user);
        List<KeyValue<String, KafkaError>> errors = dlqTopic.readKeyValuesToList();
        assertEquals("1", errors.get(0).key);
    }

    @Override
    protected Map<String, String> getSpecificProperties() {
        return Map.of(STATE_DIR_CONFIG, "/tmp/kafka-streams");
    }
}

To use Kstreamplify in a Spring Boot application, follow these steps:

1. Add the dependency: Include kstreamplify-spring-boot. You may need to override the kafka.version property in your pom.xml to ensure compatibility with the Spring Boot Starter Parent.
2. Implement the starter: Create a class that extends KafkaStreamsStarter and override the topology(StreamsBuilder streamsBuilder) method to define your business logic. You should also override dlqTopic() to specify your error topic.
3. Configure properties: Define your Kafka settings in application.yml under the kafka.properties key.

Note: The application will automatically bootstrap the Kafka Streams instance.

<!-- Dependency setup -->
<properties>
    <kafka.version>4.3.1</kafka.version>
</properties>

<dependency>
    <groupId>com.michelin</groupId>
    <artifactId>kstreamplify-spring-boot</artifactId>
    <version>${kstreamplify.version}</version>
</dependency>

<!-- Implementation -->
@Component
public class MyKafkaStreams extends KafkaStreamsStarter {
    @Override
    public void topology(StreamsBuilder streamsBuilder) {
        // Define your topology here
    }

    @Override
    public String dlqTopic() {
        return "dlq_topic";
    }
}

<!-- Configuration in application.yml -->
kafka:
  properties:
    application.id: 'myKafkaStreams'
    bootstrap.servers: 'localhost:9092'
    schema.registry.url: 'http://localhost:8081'