spring-cloud-stream-elasticsearch

The goal of this project is to implement a "News" processing pipeline composed of five Spring Boot applications: producer-api, categorizer-service, collector-service, publisher-api and news-client.

• We are using Spring Cloud Stream framework for building highly scalable event-driven applications connected with shared messaging systems.

• We are using Spring Data Elasticsearch to persist data in Elasticsearch.

• We are using Thymeleaf as HTML template

• We are using Zipkin for visualizing traces between and within applications.

• We are using Eureka, also known as Discovery Server, it is a service that holds the information about all applications.

Note	In docker-swarm-environment repository, it is shown how to deploy this project into a cluster of Docker Engines in swarm mode.

Project Architecture

Applications

producer-api

Spring Boot Web Java application that creates news and pushes news events to producer.news topic in Kafka.

categorizer-service

Spring Boot Web Java application that listens to news events in producer.news topic in Kafka, categorizes and pushes them to categorizer.news topic.

collector-service

Spring Boot Web Java application that listens for news events in categorizer.news topic in Kafka, saves them in Elasticsearch and pushes the news events to collector.news topic.

publisher-api

Spring Boot Web Java application that reads directly from Elasticsearch and exposes a REST API. It doesn’t listen from Kafka.

news-client

Spring Boot Web java application that provides a User Interface to see the news. It implements a Websocket that consumes news events from the topic collector.news. So, news are updated on the fly on the main page. Besides, news-client communicates directly with publisher-api whenever search for a specific news or news update are needed.

The Websocket operation is shown in the short gif below. A news is created in producer-api and, immediately, it is shown in news-client.

Generate NewsEvent

In a terminal and inside spring-cloud-stream-elasticsearch folder run

./mvnw clean install --projects commons-news

It will install commons-news-1.0.0.jar in you local Maven repository, so that it can be visible by all services.

Build Docker images

In a terminal and inside spring-cloud-stream-elasticsearch folder, in order to build the applications docker images, you can just run the following script

./build-apps.sh

Or manually run the following ./mvnw commands for each application.

eureka-server

./mvnw clean package dockerfile:build -DskipTests --projects eureka-server

producer-api

./mvnw clean package dockerfile:build -DskipTests --projects producer-api

Environment Variable	Description
KAFKA_HOST	Specify host of the Kafka message broker to use (default localhost)
KAFKA_PORT	Specify port of the Kafka message broker to use (default 29092)
EUREKA_HOST	Specify host of the Eureka service discovery to use (default localhost)
EUREKA_PORT	Specify port of the Eureka service discovery to use (default 8761)
ZIPKIN_HOST	Specify host of the Zipkin distributed tracing system to use (default localhost)
ZIPKIN_PORT	Specify port of the Zipkin distributed tracing system to use (default 9411)

categorizer-service

./mvnw clean package dockerfile:build -DskipTests --projects categorizer-service

Environment Variable	Description
KAFKA_HOST	Specify host of the Kafka message broker to use (default localhost)
KAFKA_PORT	Specify port of the Kafka message broker to use (default 29092)
EUREKA_HOST	Specify host of the Eureka service discovery to use (default localhost)
EUREKA_PORT	Specify port of the Eureka service discovery to use (default 8761)
ZIPKIN_HOST	Specify host of the Zipkin distributed tracing system to use (default localhost)
ZIPKIN_PORT	Specify port of the Zipkin distributed tracing system to use (default 9411)

collector-service

./mvnw clean package dockerfile:build -DskipTests --projects collector-service

Environment Variable	Description
ELASTICSEARCH_HOST	Specify host of the Elasticsearch search engine to use (default localhost)
ELASTICSEARCH_NODES_PORT	Specify nodes port of the Elasticsearch search engine to use (default 9300)
ELASTICSEARCH_REST_PORT	Specify rest port of the Elasticsearch search engine to use (default 9200)
KAFKA_HOST	Specify host of the Kafka message broker to use (default localhost)
KAFKA_PORT	Specify port of the Kafka message broker to use (default 29092)
EUREKA_HOST	Specify host of the Eureka service discovery to use (default localhost)
EUREKA_PORT	Specify port of the Eureka service discovery to use (default 8761)
ZIPKIN_HOST	Specify host of the Zipkin distributed tracing system to use (default localhost)
ZIPKIN_PORT	Specify port of the Zipkin distributed tracing system to use (default 9411)

publisher-api

./mvnw clean package dockerfile:build -DskipTests --projects publisher-api

Environment Variable	Description
ELASTICSEARCH_HOST	Specify host of the Elasticsearch search engine to use (default localhost)
ELASTICSEARCH_NODES_PORT	Specify nodes port of the Elasticsearch search engine to use (default 9300)
ELASTICSEARCH_REST_PORT	Specify rest port of the Elasticsearch search engine to use (default 9200)
EUREKA_HOST	Specify host of the Eureka service discovery to use (default localhost)
EUREKA_PORT	Specify port of the Eureka service discovery to use (default 8761)
ZIPKIN_HOST	Specify host of the Zipkin distributed tracing system to use (default localhost)
ZIPKIN_PORT	Specify port of the Zipkin distributed tracing system to use (default 9411)

news-client

./mvnw clean package dockerfile:build -DskipTests --projects news-client

Environment Variable	Description
KAFKA_HOST	Specify host of the Kafka message broker to use (default localhost)
KAFKA_PORT	Specify port of the Kafka message broker to use (default 29092)
EUREKA_HOST	Specify host of the Eureka service discovery to use (default localhost)
EUREKA_PORT	Specify port of the Eureka service discovery to use (default 8761)
ZIPKIN_HOST	Specify host of the Zipkin distributed tracing system to use (default localhost)
ZIPKIN_PORT	Specify port of the Zipkin distributed tracing system to use (default 9411)

Start Environment

Open a terminal and inside spring-cloud-stream-elasticsearch root folder run

docker-compose up -d

Wait a until all containers are Up (healthy). You can check their status by running

docker-compose ps

Running Applications as Docker containers

Open a terminal and inside spring-cloud-stream-elasticsearch root folder run following script

./start-apps.sh

Running Applications with Maven

During development, it is easier to just run the applications instead of always build the docker images and run it. For it, inside spring-cloud-stream-elasticsearch, run the following Maven commands in different terminals

eureka-server

./mvnw spring-boot:run --projects eureka-server

producer-api

./mvnw spring-boot:run --projects producer-api -Dspring-boot.run.jvmArguments="-Dserver.port=9080"

categorizer-service

./mvnw spring-boot:run --projects categorizer-service -Dspring-boot.run.jvmArguments="-Dserver.port=9081"

collector-service

./mvnw spring-boot:run --projects collector-service -Dspring-boot.run.jvmArguments="-Dserver.port=9082"

publisher-api

./mvnw spring-boot:run --projects publisher-api -Dspring-boot.run.jvmArguments="-Dserver.port=9083"

news-client

./mvnw spring-boot:run --projects news-client

Application URLs

Application	URL
producer-api	http://localhost:9080/swagger-ui.html
publisher-api	http://localhost:9083/swagger-ui.html
news-client	http://localhost:8080

Shutdown

Run the command below to stop the applications

./stop-apps.sh

Then, run the following command to stop and remove docker-compose containers, networks and volumes

docker-compose down -v

Useful links

Eureka

Eureka can be accessed at http://localhost:8761

Kafka Topics UI

Kafka Topics UI can be accessed at http://localhost:8085

Zipkin

Zipkin can be accessed at http://localhost:9411

The figure below shows an example of the complete flow a news passes through. It goes since producer-api, where the news is created, until news-client.

Kafka Manager

Kafka Manager can be accessed at http://localhost:9000

Configuration

• First, you must create a new cluster. Click on Cluster (dropdown on the header) and then on Add Cluster

• Type the name of your cluster in Cluster Name field, for example: MyZooCluster

• Type zookeeper:2181 in Cluster Zookeeper Hosts field

• Enable checkbox Poll consumer information (Not recommended for large # of consumers)

• Click on Save button at the bottom of the page.

The figure below shows the consumers os the Kafka topics. As we can see, the consumers are updated as the lag is 0

Elasticsearch REST API

Check ES is up and running

curl http://localhost:9200

Check indexes in ES

curl http://localhost:9200/_cat/indices?v

Check news index mapping

curl http://localhost:9200/news/_mapping

Simple search

curl http://localhost:9200/news/news/_search

TODO

• add alias to the index: wait for this feature be available in Spring Data Elasticsearch (https://jira.spring.io/browse/DATAES-192)

• news-client: bug. everytime sync is clicked, it enables Websocket;

• news-client: if websocket is enabled/disabled, sync button should be disabled/enabled;

• news-client: implement pagination;

Issues

• In news-client application.yml, I needed to set the property spring.sleuth.async.enabled to false otherwise the Websocket wouldn’t work. I was having the following exception

  [source]
  ----
  ERROR [news-client,39ba826690e440da,39ba826690e440da,true] 1 --- [nio-8080-exec-2] o.s.c.s.i.web.ExceptionLoggingFilter     : Uncaught exception thrown
  org.springframework.web.util.NestedServletException: Request processing failed; nested exception is org.springframework.web.socket.sockjs.SockJsException:
  Uncaught failure in SockJS request, uri=http://localhost:8080/news-websocket/202/gcbccz4x/websocket; nested exception is org.springframework.web.socket.sockjs.SockJsException:
  Uncaught failure for request http://localhost:8080/news-websocket/202/gcbccz4x/websocket; nested exception is org.springframework.core.task.TaskRejectedException:
  Executor [java.util.concurrent.ScheduledThreadPoolExecutor@2427859c[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 1]] did not accept task: org.springframework.cloud.sleuth.instrument.async.TraceRunnable@51841203
  ----