Each service within an aggregate defines a service descriptor in the repository’s services module.

A service descriptor defines the external resources a service uses and the api it exposes. The types of resources a descriptor can reference depends on the installed Creek extensions.

This demo will use the Kafka Streams extension, and the handle-occurrence-service’s descriptor will define a twitter.tweet.text input topic, which the service will consume, and a twitter.handle.usage output topic, which the service will produce to.

Note: To keep this tutorial self-contained, the service’s input topic is owned by the service. It would be more common for an upstream service or aggregate to own the topic and for the topic’s definition to be imported from there.

Define the JSON output payload type

The input topic carries raw tweet text as a Kafka-native String. The output topic uses schema-validated JSON, via the Creek Kafka JSON serde. Its Java type lives in the repository’s services module, where services in this aggregate can share the payload type.

Add the following record to services/src/main/java/io/github/creek/service/basic/kafka/streams/demo/services/model/HandleUsage.java:

@GeneratesSchema
public record HandleUsage(
        @JsonProperty(required = true)
        @Schema(minLength = 1)
        String handle,
        @Schema(minimum = "1")
        int count) {

    public HandleUsage {
        requireNonNull(handle, "handle");
        if (handle.isEmpty()) {
            throw new IllegalArgumentException("handle cannot be empty");
        }
        if (count <= 0) {
            throw new IllegalArgumentException("count must be greater than zero");
        }
    }
}

The @GeneratesSchema annotation tells Creek’s JSON schema Gradle plugin to generate a JSON schema for the type. The plugin does not infer the compact constructor’s validation rules automatically, so they’re expressed explicitly: @JsonProperty(required = true) marks a field as non-optional, and Swagger’s @Schema(minLength = ...)/@Schema(minimum = ...) annotations capture the non-empty handle and positive count constraints in the generated schema.

To allow Jackson to do its magic, as we’re running under JPMS, we need to open the model package so Jackson can access the types at runtime.

Add the following to services/src/main/java/module-info.java:

// Required so Jackson can access the JSON types at runtime.
opens io.github.creek.service.basic.kafka.streams.demo.services.model;

ProTip: Run ./gradlew :services:generateJsonSchema to generate the schema without running a full build. The generated schema file is written under services/build/generated/resources/schema/main/.

JSON Schema generation

The aggregate-template repo supplies the necessary configuration and code to make generating a JSON schema easy.

The services module’s build.gradle.kts applies the plugin, adds the required annotation dependencies, and tells it which module to scan for @GeneratesSchema annotated types:

plugins {
    `java-library`
    id("org.creekservice.schema.json")
}

dependencies {
    api(project(":api"))

    compileOnlyApi("org.creekservice:creek-base-annotation:${property("creekVersion")}")
    compileOnlyApi("com.fasterxml.jackson.core:jackson-annotations:${property("jacksonAnnotationsVersion")}")
    compileOnlyApi("io.swagger.core.v3:swagger-annotations:${property("swaggerAnnotationsVersion")}")
}

creek.schema.json {
    typeScanning.moduleWhiteList(moduleName)
    subTypeScanning.moduleWhiteList(moduleName)
}

The services module’s module-info.java includes the necessary annotations dependencies:

requires static com.fasterxml.jackson.annotation;
requires static creek.base.annotation;
requires static io.swagger.v3.oas.annotations;

Define the topic resources

The aggregate template used to bootstrap the repository provided a shell service descriptor in the repository named HandleOccurrenceServiceDescriptor.java. Add the following to the class to define the service’s input and output topics:

import static io.github.creek.service.basic.kafka.streams.demo.internal.TopicConfigBuilder.withPartitions;
import static io.github.creek.service.basic.kafka.streams.demo.internal.TopicDescriptors.JSON_FORMAT;
import static io.github.creek.service.basic.kafka.streams.demo.internal.TopicDescriptors.KAFKA_FORMAT;
import static io.github.creek.service.basic.kafka.streams.demo.internal.TopicDescriptors.inputTopic;
import static io.github.creek.service.basic.kafka.streams.demo.internal.TopicDescriptors.outputTopic;

import org.creekservice.api.kafka.metadata.topic.OwnedKafkaTopicInput;
import org.creekservice.api.kafka.metadata.topic.OwnedKafkaTopicOutput;

public final class HandleOccurrenceServiceDescriptor implements ServiceDescriptor {

    ...

    // Define the tweet-text input topic, conceptually owned by this service:
    public static final OwnedKafkaTopicInput<Long, String> RawTweetStream =
            inputTopic(
                "twitter.tweet.text", // Topic name
                Long.class, // Topic key: Tweet id (Kafka native)
                KAFKA_FORMAT,
                String.class, // Topic value: Tweet text (Kafka native)
                KAFKA_FORMAT,
                withPartitions(5)); // Topic config

    // Define the output topic, again conceptually owned by this service:
    public static final OwnedKafkaTopicOutput<String, HandleUsage> TweetHandleUsageStream =
            outputTopic(
                "twitter.handle.usage",
                String.class, // Twitter handle (Kafka native)
                KAFKA_FORMAT,
                HandleUsage.class,  // Usage data (JSON)
                JSON_FORMAT,
                withPartitions(6)
                    .withRetentionTime(Duration.ofHours(12)));

    ...
}

The two class constants define the input and output topics the services use. These constants will be used later when building the Kafka Streams topology.

Each topic definition includes the topic name, the types stored in the topic’s records’ key and value, and the topic config. Both definitions specify the serialization formats explicitly: RawTweetStream uses Kafka’s native format for its Long key and String tweet-text value, so no JSON schema is needed for the input. TweetHandleUsageStream uses a Kafka-native String key and a schema-validated JSON HandleUsage value as defined above.

Note: A topic’s JSON schema is a resource, just like the topic itself, and is owned by whichever service owns the topic. If another service later consumes this topic as an input, by calling toInput() on TweetHandleUsageStream (see the next tutorial), the schema remains owned by this service — the consuming service only gets an unowned reference to it.

In this instance, the topic config defines the number of partitions and, for one topic, the retention time for records in the topic. If no retention time was set, the cluster default would be used.

ProTip: Defaulting to the cluster’s default topic retention time can be useful as it allows different clusters to define different defaults. For example, development, QA and Staging environments can have much shorter times than production.

ProTip: The TopicConfigBuilder class, which defines the withPartitions and withRetentionTime methods used above, is part of the Git repository. It can be customised as your use-case requires.

The register method wrapping each resource descriptor ensures they are registered with the outer service descriptor.

Note: The system tests we’ll define later will use the service descriptor to discover the service metadata required to run the service, pipe in inputs and read outputs.

Updated: