Create a Streaming Application with the Oracle Cloud Infrastructure Streaming Service

This guide describes how to use the Graal Development Kit for Micronaut (GDK) to create a streaming application that uses Micronaut^® Streaming and the Oracle Cloud Infrastructure Streaming service. The application consists of two Micronaut microservices that use the Oracle Cloud Infrastructure Streaming service to communicate with each other in an asynchronous and decoupled way.

Oracle Cloud Infrastructure Streaming service is a real-time, serverless, streaming platform compatible with Apache Kafka.

Prerequisites #

• JDK 17 or higher. See Setting up Your Desktop.
• An Oracle Cloud Infrastructure account. See Setting up Your Cloud Accounts.
• The Oracle Cloud Infrastructure CLI installed with local access configured.
• An Oracle Cloud Infrastructure compartment where you will create a stream. Ensure appropriate permissions are granted to your user account to manage the Streaming service.
• The GDK CLI. See Setting up Your Desktop. (Optional.)

Follow the steps below to create the application from scratch. However, you can also download the completed example:

A note regarding your development environment

Consider using Visual Studio Code, which provides native support for developing applications with the Graal Development Kit extension.

If you use IntelliJ IDEA, enable annotation processing.

Windows platform: The GDK guides are compatible with Gradle only. Maven support is coming soon.

1. Create the Microservices #

The two microservices are:

• Books returns a list of books. It uses a domain consisting of a book name and an International Standard Book Number (ISBN). It also publishes a message to the streaming service every time a book is accessed.
• Analytics connects to the streaming service to update the analytics for every book (a counter). It also exposes an endpoint to retrieve the counter.

1.1. Create the Books Microservice #

1. Open the GDK Launcher in advanced mode.

2. Create a new project using the following selections.
   • Project Type: Application (Default)
   • Project Name: books
   • Base Package: com.example.publisher
   • Clouds: OCI
   • Build Tool: Gradle (Groovy) or Maven
   • Language: Java (Default)
   • Test Framework: JUnit (Default)
   • Java Version: 17 (Default)
   • Micronaut Version: (Default)
   • Cloud Services: Streaming
   • Features: Awaitility Framework, GraalVM Native Image, Micronaut Serialization Jackson Core, and Reactor
   • Sample Code: No
3. Click Generate Project, then click Download Zip. The GDK Launcher creates an application with the default package com.example.publisher in a directory named books. The application ZIP file will be downloaded to your default downloads directory. Unzip it, open it in your code editor, and proceed to the next steps.

Alternatively, use the GDK CLI as follows:

gdk create-app com.example.publisher.books \
    --clouds=oci \
    --services=streaming \
    --features=awaitility,graalvm,reactor,serialization-jackson \
    --build=gradle \
    --jdk=17 \
    --lang=java \
    --example-code=false

gdk create-app com.example.publisher.books \
    --clouds=oci \
    --services=streaming \
    --features=awaitility,graalvm,reactor,serialization-jackson \
    --build=maven \
    --jdk=17 \
    --lang=java \
    --example-code=false

1.1.1. Book Domain Class

The GDK Launcher created a Book domain class in a file named lib/src/main/java/com/example/publisher/Book.java, as follows:

package com.example.publisher;

import io.micronaut.core.annotation.Creator;
import io.micronaut.serde.annotation.Serdeable;

import java.util.Objects;

@Serdeable
public class Book {

    private final String isbn;
    private final String name;

    @Creator
    public Book(String isbn, String name) {
        this.isbn = isbn;
        this.name = name;
    }

    public String getIsbn() {
        return isbn;
    }

    public String getName() {
        return name;
    }

    @Override
    public String toString() {
        return "Book{" +
                "isbn='" + isbn + '\'' +
                ", name='" + name + '\'' +
                '}';
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        Book other = (Book) o;
        return Objects.equals(isbn, other.isbn) &&
                Objects.equals(name, other.name);
    }

    @Override
    public int hashCode() {
        return Objects.hash(isbn, name);
    }
}

1.1.2. BookService

To keep this guide simple there is no database persistence: the Books microservice keeps the list of books in memory. The GDK Launcher created a class named BookService in lib/src/main/java/com/example/publisher/BookService.java with the following contents:

package com.example.publisher;

import jakarta.annotation.PostConstruct;
import jakarta.inject.Singleton;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

@Singleton
public class BookService {

    private final List<Book> bookStore = new ArrayList<>();

    @PostConstruct
    void init() {
        bookStore.add(new Book("1491950358", "Building Microservices"));
        bookStore.add(new Book("1680502395", "Release It!"));
        bookStore.add(new Book("0321601912", "Continuous Delivery"));
    }

    public List<Book> listAll() {
        return bookStore;
    }

    public Optional<Book> findByIsbn(String isbn) {
        return bookStore.stream()
                .filter(b -> b.getIsbn().equals(isbn))
                .findFirst();
    }
}

1.1.3. BookController

The GDK Launcher created a controller to access Book instances in a file named lib/src/main/java/com/example/publisher/BookController.java with the following contents:

package com.example.publisher;

import io.micronaut.http.annotation.Controller;
import io.micronaut.http.annotation.Get;

import java.util.List;
import java.util.Optional;

@Controller("/books") // <1>
class BookController {

    private final BookService bookService;

    BookController(BookService bookService) { // <2>
        this.bookService = bookService;
    }

    @Get // <3>
    List<Book> listAll() {
        return bookService.listAll();
    }

    @Get("/{isbn}") // <4>
    Optional<Book> findBook(String isbn) {
        return bookService.findByIsbn(isbn);
    }
}

1 The @Controller annotation defines the class as a controller mapped to the root URI /books.

2 Use constructor injection to inject a bean of type BookService.

3 The @Get annotation maps the listAll method to an HTTP GET request on /books.

4 The @Get annotation maps the findBook method to an HTTP GET request on /books/{isbn}.

1.1.4. AnalyticsClient

The GDK Launcher created a client interface to send messages to the streaming service in a file named lib/src/main/java/com/example/publisher/AnalyticsClient.java with the contents shown below. (Micronaut generates an implementation for the client interface at compilation time.)

package com.example.publisher;

import io.micronaut.configuration.kafka.annotation.KafkaClient;
import io.micronaut.configuration.kafka.annotation.Topic;
import reactor.core.publisher.Mono;

@KafkaClient
public interface AnalyticsClient {

    @Topic("analytics") // <1>
    Mono<Book> updateAnalytics(Book book); // <2>
}

1 Set the name of the topic. This matches the name of the topic used by AnalyticsListener in the Analytics microservice.

Note: This must match the name of the topic that you will create later.

2 Send the Book POJO. Micronaut will automatically convert it to JSON before sending it.

1.1.5. AnalyticsFilter

Sending a message to the streaming service is as simple as injecting AnalyticsClient and calling its updateAnalytics method. The goal is to send a message every time the details of a book are returned from the Books microservice or, in other words, every time there is a call to http://localhost:8080/books/{isbn}. To achieve this, the GDK Launcher created an Http Server Filter in a file named lib/src/main/java/com/example/publisher/AnalyticsFilter.java as follows:

package com.example.publisher;

import io.micronaut.http.HttpRequest;
import io.micronaut.http.MutableHttpResponse;
import io.micronaut.http.annotation.Filter;
import io.micronaut.http.filter.HttpServerFilter;
import io.micronaut.http.filter.ServerFilterChain;
import reactor.core.publisher.Flux;
import org.reactivestreams.Publisher;

@Filter("/books/?*") // <1>
class AnalyticsFilter implements HttpServerFilter { // <2>

    private final AnalyticsClient analyticsClient;

    AnalyticsFilter(AnalyticsClient analyticsClient) { // <3>
        this.analyticsClient = analyticsClient;
    }

    @Override
    public Publisher<MutableHttpResponse<?>> doFilter(HttpRequest<?> request,
                                                      ServerFilterChain chain) { // <4>
        return Flux
                .from(chain.proceed(request)) // <5>
                .flatMap(response -> {
                    Book book = response.getBody(Book.class).orElse(null); // <6>
                    if (book == null) {
                        return Flux.just(response);
                    }
                    return Flux.from(analyticsClient.updateAnalytics(book)).map(b -> response); // <7>
                });
    }
}

1 Annotate the class with @Filter and define the Ant-style matcher pattern to intercept all calls to the desired URIs.

2 The class must implement HttpServerFilter.

3 Dependency injection for AnalyticsClient.

4 Implement the doFilter method.

5 Call the request; this will invoke the controller action.

6 Get the response from the controller and return the body as an instance of the Book class.

7 If the book is retrieved, use the client to send a message.

1.2. Create the Analytics Microservice #

1. Open the GDK Launcher in advanced mode.

2. Create a new project using the following selections.
   • Project Type: Application (Default)
   • Project Name: analytics
   • Base Package: com.example.consumer
   • Clouds: OCI
   • Build Tool: Gradle (Groovy) or Maven
   • Language: Java (Default)
   • Test Framework: JUnit (Default)
   • Java Version: 17 (Default)
   • Micronaut Version: (Default)
   • Cloud Services: Streaming
   • Features: Awaitility Framework, GraalVM Native Image and Micronaut Serialization Jackson Core
   • Sample Code: No
3. Click Generate Project, then click Download Zip. The GDK Launcher creates an application with the default package com.example.consumer in a directory named analytics. The application ZIP file will be downloaded to your default downloads directory. Unzip it, open it in your code editor, and proceed to the next steps.

Alternatively, use the GDK CLI as follows:

gdk create-app com.example.consumer.analytics \
    --clouds=oci \
    --services=streaming \
    --features=awaitility,graalvm,serialization-jackson \
    --build=gradle \
    --jdk=17 \
    --lang=java \
    --example-code=false

gdk create-app com.example.consumer.analytics \
    --clouds=oci \
    --services=streaming \
    --features=awaitility,graalvm,serialization-jackson \
    --build=maven \
    --jdk=17 \
    --lang=java \
    --example-code=false

1.2.1. Domain Classes

The GDK Launcher created a Book domain class in a file named lib/src/main/java/com/example/consumer/Book.java, as shown below. (This Book POJO is the same as the one in the Books microservice. In a real application this would be in a shared library.)

package com.example.consumer;

import io.micronaut.core.annotation.Creator;
import io.micronaut.serde.annotation.Serdeable;

import java.util.Objects;

@Serdeable
public class Book {

    private final String isbn;
    private final String name;

    @Creator
    public Book(String isbn, String name) {
        this.isbn = isbn;
        this.name = name;
    }

    public String getIsbn() {
        return isbn;
    }

    public String getName() {
        return name;
    }

    @Override
    public String toString() {
        return "Book{" +
                "isbn='" + isbn + '\'' +
                ", name='" + name + '\'' +
                '}';
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        Book other = (Book) o;
        return Objects.equals(isbn, other.isbn) &&
                Objects.equals(name, other.name);
    }

    @Override
    public int hashCode() {
        return Objects.hash(isbn, name);
    }
}

The GDK Launcher also created a BookAnalytics domain class in a file named lib/src/main/java/com/example/consumer/BookAnalytics.java, as follows:

package com.example.consumer;

import io.micronaut.core.annotation.Creator;
import io.micronaut.serde.annotation.Serdeable;

@Serdeable
public class BookAnalytics {

    private final String bookIsbn;
    private final long count;

    @Creator
    public BookAnalytics(String bookIsbn, long count) {
        this.bookIsbn = bookIsbn;
        this.count = count;
    }

    public String getBookIsbn() {
        return bookIsbn;
    }

    public long getCount() {
        return count;
    }
}

1.2.2. AnalyticsService

To keep this guide simple there is no database persistence: the Analytics microservice keeps book analytics in memory. The GDK Launcher created a class named AnalyticsService in lib/src/main/java/com/example/consumer/AnalyticsService.java with the following contents:

package com.example.consumer;

import jakarta.inject.Singleton;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

@Singleton
public class AnalyticsService {

    private final Map<Book, Long> bookAnalytics = new ConcurrentHashMap<>(); // <1>

    public void updateBookAnalytics(Book book) { // <2>
        bookAnalytics.compute(book, (k, v) -> {
            return v == null ? 1L : v + 1;
        });
    }

    public List<BookAnalytics> listAnalytics() { // <3>
        return bookAnalytics
                .entrySet()
                .stream()
                .map(e -> new BookAnalytics(e.getKey().getIsbn(), e.getValue()))
                .collect(Collectors.toList());
    }
}

1 Keep the book analytics in memory.

2 Initialize and update the analytics for the book passed as parameter.

3 Return all the analytics.

1.2.3. AnalyticsController

The GDK Launcher created a Controller to create an endpoint for the Analytics microservice in a file named lib/src/main/java/com/example/consumer/AnalyticsController.java, as follows:

package com.example.consumer;

import io.micronaut.http.annotation.Controller;
import io.micronaut.http.annotation.Get;

import java.util.List;

@Controller("/analytics") // <1>
class AnalyticsController {

    private final AnalyticsService analyticsService;

    AnalyticsController(AnalyticsService analyticsService) {
        this.analyticsService = analyticsService;
    }

    @Get // <2>
    List<BookAnalytics> listAnalytics() {
        return analyticsService.listAnalytics();
    }
}

1 The @Controller annotation defines the class as a controller mapped to the root URI /analytics.

2 The @Get annotation maps the listAnalytics method to an HTTP GET request on /analytics.

The application doesn’t expose the method updateBookAnalytics created in AnalyticsService. This method will be invoked when reading messages from Kafka.

1.2.4. AnalyticsListener

The GDK Launcher created a class to act as a consumer of the messages sent to the streaming service by the Books microservice. The Micronaut framework implements logic to invoke the consumer at compile time. The AnalyticsListener class is in a file named lib/src/main/java/com/example/consumer/AnalyticsListener.java, as follows:

package com.example.consumer;

import io.micronaut.configuration.kafka.annotation.KafkaListener;
import io.micronaut.configuration.kafka.annotation.Topic;
import io.micronaut.context.annotation.Requires;
import io.micronaut.context.env.Environment;

@Requires(notEnv = Environment.TEST) // <1>
@KafkaListener // <2>
class AnalyticsListener {

    private final AnalyticsService analyticsService; // <3>

    AnalyticsListener(AnalyticsService analyticsService) { // <3>
        this.analyticsService = analyticsService;
    }

    @Topic("analytics") // <4>
    void updateAnalytics(Book book) {
        analyticsService.updateBookAnalytics(book); // <5>
    }
}

1 Do not load this bean in the test environment: you can run tests without access to a streaming service.

2 Annotate the class with @KafkaListener to indicate that this bean consumes messages from Kafka.

3 Constructor injection for AnalyticsService.

4 Annotate the method with @Topic and specify the topic name. This matches the name of the topic used by AnalyticsClient in the Books microservice.

Note: This must match the name of the topic that you will create later.

5 Call AnalyticsService to update the analytics for the book.

1.2.5. Change the port for the Analytics Microservice

The Books and Analytics microservices are both run on your local machine, so they must run on different ports. Change the port that Analytics runs on by editing its oci/src/main/resources/application.properties to include the following configuration:

micronaut.server.port=8081

2. Use Streams with Micronaut #

The Micronaut Oracle Cloud module provides integration between Micronaut applications and Oracle Cloud Infrastructure services, including Streams. The GDK Launcher added the appropriate dependencies to your build file when you selected the Streaming service in the launcher.

The Micronaut application configuration file contains the properties to authenticate your microservices against Oracle Cloud Infrastructure.

Modify the file named oci/src/main/resources/application-oraclecloud.properties for each microservice to include the following:

oci.config.profile=${OCI_PROFILE}

3. Set Up Oracle Cloud Infrastructure Resources #

Each microservice’s oci/src/main/resources/application-oraclecloud.properties file relies on the values of environment variables to provide connection details. For example, the Books microservice has the following contents:

oci.config.profile=${OCI_PROFILE}
kafka.bootstrap.servers=${OCI_STREAM_POOL_FQDN}:9092
kafka.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="${OCI_TENANCY_NAME}/${OCI_USERNAME}/${OCI_STREAM_POOL_OCID}" password="${OCI_AUTH_TOKEN}";
kafka.security.protocol=SASL_SSL

In this section, you provide the values for those environment variables by using the Oracle Cloud Infrastructure Console to create an authentication token, then create and configure a stream.

3.1. Set the Value of OCI_USERNAME #

1. Sign in to the Oracle Cloud Console, click your user avatar , and then select the user name.

2. Copy the user name, and set the value of the OCI_USERNAME environment variable, for example:

    export OCI_USERNAME=john.doe@oracle.com
   

3.2. Set the Value of OCI_TENANCY_NAME #

1. From the Oracle Cloud Console, click your user avatar , and then select the tenancy.

2. Copy the tenancy name, and set the value of the OCI_TENANCY_NAME environment variable, for example:

    export OCI_TENANCY_NAME=GdkDeveloper
   

3.3. Set the Value of OCI_AUTH_TOKEN #

1. From the Oracle Cloud Console, click your user avatar , and then select User settings.

2. Follow the steps to create an auth token. Provide a description for the token, for example “gdk-stream-token”.

3. Set the value of the OCI_AUTH_TOKEN environment variable to be the token string you created above (enclosed in double quotes), for example:

    export OCI_AUTH_TOKEN="MQ)a:)mI{<Xz26<MJh7f"
   

3.4. Create a Stream #

1. Follow the steps to create a stream, using the following properties:

   • Stream name: Enter the name for the stream: “analytics”.

     Note: the name of the stream must match the name of the topic in AnalyticsClient and AnalyticsListener.

   • Compartment: Select a compartment from the drop-down list.
   • Use the defaults for the remaining settings, then click Create.

2. When the stream is active, click the name of the stream pool in which you created the stream. (If you did not create a new stream pool, its name will be “DefaultPool”.)

3. In the “Stream Pool Information” tab, copy its OCID (click Copy) and set it as the value of the OCI_STREAM_POOL_OCID environment variable, for example:

    export OCI_STREAM_POOL_OCID=ocid1.streampool.oc1.phx.amaaaaaac...dt7yc6bv3uw533a
   

4. Copy the value of its FQDN, and set it as the value of the OCI_STREAM_POOL_FQDN environment variable, for example:

    export OCI_STREAM_POOL_FQDN=cell-1.streaming.us-phoenix-1.oci.oraclecloud.com
   

5. If you haven’t created an API-key for your Oracle Cloud Infrastructure region, run the following command:

    oci setup bootstrap
   

   The command opens your browser for authentication. When prompted, provide the tenancy’s region and a name for the profile, for example, “oci_stream_demo_profile”.

   Save the name of the profile as the value of the OCI_PROFILE environment variable, as follows:

    export OCI_PROFILE=oci_stream_demo_profile
   

   Note: For more information, see Oracle Cloud Infrastructure SDK Authentication Methods and Setting up the Configuration File.

4. Run the Application #

Your application consists of two microservices, so start each one.

4.1. Start the Books Microservice #

To run the Books microservice, use the following command, which starts the application on port 8080.

MICRONAUT_ENVIRONMENTS=oraclecloud ./gradlew :oci:run

cmd /C "set MICRONAUT_ENVIRONMENTS=oraclecloud && gradlew :oci:run"

./mvnw install -pl lib -am
MICRONAUT_ENVIRONMENTS=oraclecloud ./mvnw -pl oci mn:run

mvnw install -pl lib -am
cmd /C "set MICRONAUT_ENVIRONMENTS=oraclecloud && mvnw -pl oci mn:run"

4.2. Start the Analytics Microservice #

To run the Analytics microservice, use the following command, which starts the application on port 8081.

MICRONAUT_ENVIRONMENTS=oraclecloud ./gradlew :oci:run

cmd /C "set MICRONAUT_ENVIRONMENTS=oraclecloud && gradlew :oci:run"

./mvnw install -pl lib -am
MICRONAUT_ENVIRONMENTS=oraclecloud ./mvnw -pl oci mn:run

mvnw install -pl lib -am
cmd /C "set MICRONAUT_ENVIRONMENTS=oraclecloud && mvnw -pl oci mn:run"

4.3. Test the Microservices #

Use curl to test the microservices, as follows.

1. Retrieve the list of books:

    curl http://localhost:8080/books
   

    [{"isbn":"1491950358","name":"Building Microservices"},{"isbn":"1680502395","name":"Release It!"},{"isbn":"0321601912","name":"Continuous Delivery"}]
   

2. Retrieve the details of a specified book:

    curl http://localhost:8080/books/1491950358
   

    {"isbn":"1491950358","name":"Building Microservices"}
   

3. Retrieve the analytics:

    curl http://localhost:8081/analytics
   

    [{"bookIsbn":"1491950358","count":1}]
   

Update the curl command to the Books microservice to retrieve other books and repeat the invocations, then re-run the curl command to the Analytics microservice to see that the counts increase.

5. Generate a Native Executable Using GraalVM #

The GDK supports compiling Java applications ahead-of-time into native executables using GraalVM Native Image. You can use the Gradle plugin for GraalVM Native Image building/Maven plugin for GraalVM Native Image building. Packaged as a native executable, it significantly reduces application startup time and memory footprint.

Prerequisites: Make sure you have installed a GraalVM JDK. The easiest way to get started is with SDKMAN!. For other installation options, visit the Downloads section.

To generate a native executable, run the following command for each microservice:

./gradlew :oci:nativeCompile

MICRONAUT_ENVIRONMENTS=oraclecloud oci/build/native/nativeCompile/oci

gradlew :oci:nativeCompile

cmd /C "set MICRONAUT_ENVIRONMENTS=oraclecloud && oci\build\native\nativeCompile\oci"

./mvnw install -pl lib -am
./mvnw clean package -pl oci -Dpackaging=native-image

MICRONAUT_ENVIRONMENTS=oraclecloud oci/target/oci

mvnw install -pl lib -am
mvnw clean package -pl oci -Dpackaging=native-image

cmd /C "set MICRONAUT_ENVIRONMENTS=oraclecloud && oci\target\oci"

Start the native executable for each microservice and run the same curl requests as before to check that everything works as expected.

You can see that the microservices behave identically as if you run them as Java applications, but with reduced startup time and smaller memory footprint.

Summary #

In this guide you created a streaming application with the Micronaut framework and Oracle Cloud Infrastructure Streaming service. The communication between two microservices acting as a producer and consumer ran asynchronously. Then you packaged these microservices into native executables with GraalVM Native Image for their faster startup and reduced memory footprint.

Related Documentation #

• Oracle Cloud Infrastructure Streaming service
• Securely Store Application Secrets in Oracle Cloud Infrastructure Vault
• Kafka support in Micronaut