Event Driven Microservices with Spring Cloud Stream #jjug_ccc #ccc_ab3

‹#›© 2016 Pivotal Software, Inc. All rights reserved. ‹#›© 2016 Pivotal Software, Inc. All rights reserved. Event Driven Microservices with Spring Cloud Stream Toshiaki Maki (@making) 2016-12-03 JJUG CCC 2016 Fall #jjug_ccc #ccc_ab3 http://bit.ly/making_ccc_a3 (source code)

© 2016 Pivotal Software, Inc. All rights reserved. Spring Cloud Stream • Event-driven microservice framework • Built on battle-tested components (Spring Boot / Spring Integration) • Opinionated primitives for streaming applications • Persistent Pub/Sub • Consumer Groups • Partitioning Support • Pluggable messaging middleware bindings source | processor | sink

© 2016 Pivotal Software, Inc. All rights reserved. Spring Cloud Stream Applications Twitter Stream Cassandra java -jar twittersource.jar --server.port=8080 --consumerKey=XYZ --consumerSecret=ABC --spring.cloud.stream.bindings. output.destination=ingest Source Sink java -jar cassandrasink.jar --server.port=8081 --spring.cassandra.keyspace=tweet --spring.cloud.stream.bindings. input.destination=ingest

© 2016 Pivotal Software, Inc. All rights reserved. Spring Cloud Stream Applications Twitter Stream Cassandra java -jar twittersource.jar --server.port=8080 --consumerKey=XYZ --consumerSecret=ABC --spring.cloud.stream.bindings. output.destination=ingest Source Sink java -jar cassandrasink.jar --server.port=8081 --spring.cassandra.keyspace=tweet --spring.cloud.stream.bindings. input.destination=ingest Twitter Stream Cassandraingest

© 2016 Pivotal Software, Inc. All rights reserved. Programming Model (Sink) @SpringBootApplication  @EnableBinding(Sink.class)  public class DemoSinkApp { @StreamListener(Sink.INPUT)  void receive(Message<String> message) {  System.out.println("Received " + message.getPayload());  }  public static void main(String[] args) {  SpringApplication.run(DemoSinkApp.class, args);  }  }

© 2016 Pivotal Software, Inc. All rights reserved. Programming Model (Sink) @SpringBootApplication  @EnableBinding(Sink.class)  public class DemoSinkApp { @StreamListener(Sink.INPUT)  void receive(Message<String> message) {  System.out.println("Received " + message.getPayload());  }  public static void main(String[] args) {  SpringApplication.run(DemoSinkApp.class, args);  }  } public interface Sink { String INPUT = "input"; @Input(Sink.INPUT) SubscribableChannel input();  }

© 2016 Pivotal Software, Inc. All rights reserved. Programming Model (Source) @SpringBootApplication @RestController @EnableBinding(Source.class)  public class DemoSourceApp {  @Autowired @Output(Source.OUTPUT) MessageChannel output; @GetMapping void send(@RequestParam String text) {  output.send(MessageBuilder.withPayload(text).build());  }  public static void main(String[] args) {  SpringApplication.run(DemoSourceApp.class, args);  }  }

© 2016 Pivotal Software, Inc. All rights reserved. Programming Model (Source) @SpringBootApplication @RestController @EnableBinding(Source.class)  public class DemoSourceApp {  @Autowired @Output(Source.OUTPUT) MessageChannel output; @GetMapping void send(@RequestParam String text) {  output.send(MessageBuilder.withPayload(text).build());  }  public static void main(String[] args) {  SpringApplication.run(DemoSourceApp.class, args);  }  } public interface Source { String OUTPUT = "output"; @Output(Source.OUTPUT) MessageChannel output();  }

© 2016 Pivotal Software, Inc. All rights reserved. Programming Model (Source) @SpringBootApplication @RestController @EnableBinding(Source.class)  public class DemoSourceApp {  @Autowired Source source; @GetMapping void send(@RequestParam String text) {  source.output() .send(MessageBuilder.withPayload(text).build());  }  public static void main(String[] args) {  SpringApplication.run(DemoSourceApp.class, args);  }  }

© 2016 Pivotal Software, Inc. All rights reserved. Properties (Source) spring.cloud.stream.bindings.output.destination=demo-strm spring.cloud.stream.bindings.output.contentType=applicati on/json demo- strm Source output

© 2016 Pivotal Software, Inc. All rights reserved. Binder (RabbitMQ) <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-stream-rabbit</artifactId> </dependency> demo- strm

© 2016 Pivotal Software, Inc. All rights reserved. Binder (Apache Kafka) <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-stream-kafka</artifactId> </dependency> demo- strm

© 2016 Pivotal Software, Inc. All rights reserved. Programming Model (Processor) @SpringBootApplication  @EnableBinding(Processor.class)  public class DemoProcessorApp { @StreamListener(Processor.INPUT) @SendTo(Processor.OUTPUT)  void receive(String text) {  return "[[" + text + "]]";  }  public static void main(String[] args) {  SpringApplication.run(DemoProcessorApp.class, args);  }  }

© 2016 Pivotal Software, Inc. All rights reserved. Properties (Processor) spring.cloud.stream.bindings.output.destination=my-source spring.cloud.stream.bindings.input.destination=my-source spring.cloud.stream.bindings.output.destination=my-proc spring.cloud.stream.bindings.input.destination=my-proc Source Processor Sink

© 2016 Pivotal Software, Inc. All rights reserved. Reactive API Support by Reactor @SpringBootApplication  @EnableBinding(Processor.class)  public class DemoProcessorRxApp { @StreamListener @Output(Processor.OUTPUT)  public Flux<String> receive(@Input(Processor.INPUT) Flux<String> stream) {  return stream.map(text -> "[[" + text + "]]");  }  public static void main(String[] args) {  SpringApplication.run(DemoProcessorRxApp.class, args);  }  }

© 2016 Pivotal Software, Inc. All rights reserved. Reactive API Support by Reactor @StreamListener @Output(Processor.OUTPUT)  public Flux<AverageData> receive(@Input(Processor.INPUT) Flux<SensorData> stream) {  return stream.window(Duration.ofSecond(20), Duration.ofSecond(10)) .flatMap(win -> win.groupBy(sensor -> sensor.id)) .flatMap(group -> calcAverage(group));  }

© 2016 Pivotal Software, Inc. All rights reserved. Persistent Pub-Sub HTTP Average HDFS Top N Fault Detection s1.http s1.ave Message Broker {"id":1, "temperature":38} {"id":1, "temperature":38} {"id":1, "temperature":38}

© 2016 Pivotal Software, Inc. All rights reserved. Persistent Pub-Sub HTTP Average HDFS Top N Fault Detection s1.http s1.ave Message Broker {"id":1, "temperature": {"id":1, "temperature":38} {"id":1, "temperature":38}

© 2016 Pivotal Software, Inc. All rights reserved. HTTP Average HDFS Top N Fault Detection s1.http s1.ave Message Broker {"id":1, "temperature":38} {"id":1, "temperature":38} Average Average HDFS HDFS {"id":1, "temperature":38} {"id":1, "temperature":38} {"id":1, "temperature":38} {"id":1, "temperature":38}

© 2016 Pivotal Software, Inc. All rights reserved. HTTP Average HDFS Top N Fault Detection s1.http s1.ave Message Broker {"id":1, "temperature":38} {"id":1, "temperature":38} Average Average HDFS HDFS {"id":1, "temperature":38} {"id":1, "temperature":38} {"id":1, "temperature":38} {"id":1, "temperature":38} 😩

© 2016 Pivotal Software, Inc. All rights reserved. Consumer Group HTTP Average HDFS Top N Fault Detection s1.http s1.ave Message Broker Average Average HDFS HDFS group=ave group=hdfs {"id":1, "temperature":38}

© 2016 Pivotal Software, Inc. All rights reserved. Consumer Group HTTP Average HDFS Top N Fault Detection s1.http s1.ave Message Broker Average Average HDFS HDFS group=ave group=hdfs {"id":1, "temperature":38}{"id":1, "temperature":38}

© 2016 Pivotal Software, Inc. All rights reserved. Consumer Group HTTP Average HDFS Top N Fault Detection s1.http s1.ave Message Broker Average Average HDFS HDFS group=ave group=hdfs {"id":1, "temperature":38} {"id":1, "temperature":38}

© 2016 Pivotal Software, Inc. All rights reserved. Consumer Group HTTP Average HDFS Top N Fault Detection s1.http s1.ave Message Broker Average Average HDFS HDFS group=ave group=hdfs {"id":1, "temperature":38} {"id":1, "temperature":38} 🤓

© 2016 Pivotal Software, Inc. All rights reserved. Consumer Group HTTP Average HDFS Top N Fault Detection s1.http s1.ave Message Broker Average Average HDFS HDFS group=ave group=hdfs {"id":1, "temperature":38} {"id":1, "temperature":38} 🤓 consumer group subscriptions are durable 😁

© 2016 Pivotal Software, Inc. All rights reserved. Customer Service Email Service Point Service Post Service Source Sink spring.cloud.stream.bindings.output.destination=customer spring.cloud.stream.bindings.output.contentType=applicatio n/json

© 2016 Pivotal Software, Inc. All rights reserved. Customer Service Email Service Point Service Post Service Subscribe Source Sink group=point-service group=email-service group=post-service spring.cloud.stream.bindings.input.destination=customer spring.cloud.stream.bindings.input.group=point-service

© 2016 Pivotal Software, Inc. All rights reserved. Customer Service Email Service Point Service Post Service Subscribe Source Sink group=point-service group=email-service group=post-service spring.cloud.stream.bindings.input.destination=customer spring.cloud.stream.bindings.input.group=point-service spring.cloud.stream.bindings.input.destination=customer spring.cloud.stream.bindings.input.group=email-service

© 2016 Pivotal Software, Inc. All rights reserved. Customer Service Email Service Point Service Post Service Subscribe Source Sink group=point-service group=email-service group=post-service spring.cloud.stream.bindings.input.destination=customer spring.cloud.stream.bindings.input.group=point-service spring.cloud.stream.bindings.input.destination=customer spring.cloud.stream.bindings.input.group=email-service spring.cloud.stream.bindings.input.destination=customer spring.cloud.stream.bindings.input.group=post-service

© 2016 Pivotal Software, Inc. All rights reserved. HTTP Average s1.http Average {"id":1, "temperature":38} {"id":2, "temperature":41} {"id":2, "temperature":42} {"id":1, "temperature":37} 😩 Partitioning Support(Stateful Stream)

© 2016 Pivotal Software, Inc. All rights reserved. HTTP Average s1.http Average {"id":1, "temperature":38} {"id":2, "temperature":41} {"id":2, "temperature":42} {"id":1, "temperature":37} 😩 spring.cloud.stream.bindings.<channelName>.producer.par titionKeyExpression=payload.id Partitioning Support(Stateful Stream)

© 2016 Pivotal Software, Inc. All rights reserved. Partitioning Support(Stateful Stream) HTTP Average s1.http Average {"id":1, "temperature":38} {"id":2, "temperature":41} {"id":2, "temperature":42} {"id":1, "temperature":37}

© 2016 Pivotal Software, Inc. All rights reserved. Partitioning Support(Stateful Stream) HTTP Average s1.http Average {"id":1, "temperature":38} {"id":2, "temperature":41} {"id":2, "temperature":42} {"id":1, "temperature":37} 🤓

© 2016 Pivotal Software, Inc. All rights reserved. Test Support <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-stream-test-support</artifactId> <scope>test</scope> </dependency>

© 2016 Pivotal Software, Inc. All rights reserved. Unit Test (Sink) @RunWith(SpringRunner.class)  @SpringBootTest(webEnvironment = WebEnvironment.NONE)  public class DemoSinkAppTest { @Autowired Sink sink; @Rule public OutputCapture capture = new OutputCapture(); @Test public void testReceive() { sink.input() .send(MessageBuilder.withPayload("foo").build()); assertThat(capture.toString()) .isEqualsTo("Received foo");  } }

© 2016 Pivotal Software, Inc. All rights reserved. Unit Test (Source) @RunWith(SpringRunner.class)  @SpringBootTest(webEnvironment = WebEnvironment.NONE)  public class DemoSourceAppTest { @Autowired DemoSourceApp app; @Autowired MessageCollector collector; @Autowired Source source; @Test public void testSend() { app.send("foo"); Message<String> message = collector .forChannel(source.output()).poll(); assertThat(message.getPayload()).isEqualsTo("foo");   }}

© 2016 Pivotal Software, Inc. All rights reserved. public interface CustomerEventSource { String CREATE_OUTPUT = "create-output"; String DELETE_OUTPUT = "delete-output"; @Output(CustomerEventSource.CREATE_OUTPUT) MessageChannel createOutput(); @Output(CustomerEventSource.DELETE_OUTPUT) MessageChannel deleteOutput();  } Multi Bindings

© 2016 Pivotal Software, Inc. All rights reserved. @Component  public class CustomerService {  @Autowired CustomerEventSource source; public void create(...) { source.createOutput().send(...); } public void delete() { source.deleteOutput().send(...); }  } @SpringBootApplication @EnableBinding(CustomerEventSource.class)  public class CustomerServiceApplication { /* ... */ }

© 2016 Pivotal Software, Inc. All rights reserved. @Component  public class CustomerService {  @Autowired CustomerEventSource source; public void create(...) { source.createOutput().send(...); } public void delete() { source.deleteOutput().send(...); }  } @SpringBootApplication @EnableBinding(CustomerEventSource.class)  public class CustomerServiceApplication { /* ... */ } spring.cloud.stream.bindings.create-output.destination =create spring.cloud.stream.bindings.delete-output.destination =delete

© 2016 Pivotal Software, Inc. All rights reserved. public interface CustomerEventSink { String CREATE_INPUT = "create-input"; String DELETE_INPUT = "delete-input"; @Input(CustomerEventSink.CREATE_INPUT) SubscribableChannel createInput(); @Input(CustomerEventSink.DELETE_INPUT) SubscribableChannel deleteInput();  } Multi Bindings

© 2016 Pivotal Software, Inc. All rights reserved. @Component  public class PointService { @StreamListener(CustomerEventSink.CREATE_INPUT) public void handleCreate(CustomerCreateEvent event) { } @StreamListener(CustomerEventSink.DELETE_INPUT) public void handleDelete(CustomerDeleteEvent event) { }  } @SpringBootApplication @EnableBinding(CustomerEventSink.class)  public class PointServiceApplication { /* ... */ }

© 2016 Pivotal Software, Inc. All rights reserved. @Component  public class PointService { @StreamListener(CustomerEventSink.CREATE_INPUT) public void handleCreate(CustomerCreateEvent event) { } @StreamListener(CustomerEventSink.DELETE_INPUT) public void handleDelete(CustomerDeleteEvent event) { }  } @SpringBootApplication @EnableBinding(CustomerEventSink.class)  public class PointServiceApplication { /* ... */ } spring.cloud.stream.bindings.create-input.destination =create spring.cloud.stream.bindings.create-input.group =point-service spring.cloud.stream.bindings.delete-intput.destination =delete spring.cloud.stream.bindings.delete-input.group =point-service

© 2016 Pivotal Software, Inc. All rights reserved. Spring Cloud Sleuth • Distributed tracing solution for Spring Cloud • Interactions with external systems should be instrumented automatically • Capture data simply in logs, or by sending it to Zipkin via RestTemplate / Spring Cloud Stream / ...

© 2016 Pivotal Software, Inc. All rights reserved. Spring Cloud Sleuth Stream <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-stream-slueth</artifactId> </dependency>

© 2016 Pivotal Software, Inc. All rights reserved. Zipkin Stream Server <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-sleuth-zipkin-stream</artifactId> </dependency>

© 2016 Pivotal Software, Inc. All rights reserved. Zipkin Stream Server <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-sleuth-zipkin-stream</artifactId> </dependency> @SpringBootApplication  @EnableZipkinStreamServer  public class ZipkinStreamServer {  public static void main(String[] args) {  SpringApplication.run(DemoSinkApp.class, args);  }  }

© 2016 Pivotal Software, Inc. All rights reserved. Error Handling •Depends on the message binder implementation •(Ex.) RabbitMQ binder routes the failed message to the Dead-Letter Queue(DLQ). No mechanism to handle DLQs.

© 2016 Pivotal Software, Inc. All rights reserved. Dead-Letter Queue Processing demo-strm Topic Exchange demo-strm .ave Source Sink DLX RabbitMQBinder spring.cloud.stream.bindings.input.destination=demo-strm spring.cloud.stream.bindings.input.group=ave spring.cloud.stream.rabbit.bindings.input.consumer.autoBi ndDlq=true

© 2016 Pivotal Software, Inc. All rights reserved. Dead-Letter Queue Processing demo-strm Topic Exchange demo-strm .ave Source Sink DLX demo-strm .ave.dlq RabbitMQBinder spring.cloud.stream.bindings.input.destination=demo-strm spring.cloud.stream.bindings.input.group=ave spring.cloud.stream.rabbit.bindings.input.consumer.autoBi ndDlq=true

© 2016 Pivotal Software, Inc. All rights reserved. Handling DLQ @Component  public class DlqHander { @RabbitListener(queues = "customer.email-service.dlq") public void handle(Message event) { // Re-deliver if you want }  }

© 2016 Pivotal Software, Inc. All rights reserved. Consumer Driven Contracts •Consumer shares "expectation" with Producer via "Contract" (≈ DSL) •The contract violation should be detected by generated tests on the producer side.

© 2016 Pivotal Software, Inc. All rights reserved. Consumer Driven Contracts •Consumer shares "expectation" with Producer via "Contract" (≈ DSL) •The contract violation should be detected by generated tests on the producer side. ContractConsumer Producer

© 2016 Pivotal Software, Inc. All rights reserved. Spring Cloud Contract •A CDC Solution for JVM apps (especially Spring) •Contract DSL using Groovy •Generates Acceptance test (JUnit or Spock) for producer •Generates Stub for consumer •WireMock Support for REST Test •Messaging Support (Spring Integration, Spring Cloud Stream and Apache Camel)

© 2016 Pivotal Software, Inc. All rights reserved. Contract DSL Contract.make { label 'create-customer' input { } outputMessage { sentTo('demo-strm') headers({header('Content-Type':'...')})  body('''{"name":"@making"}''') }} shouldCreateCustomer.groovy

© 2016 Pivotal Software, Inc. All rights reserved. Contract DSL Contract.make { label 'create-customer' input { } outputMessage { sentTo('demo-strm') headers({header('Content-Type':'...')})  body('''{"name":"@making"}''') }} shouldCreateCustomer.groovy Created by Consumer

© 2016 Pivotal Software, Inc. All rights reserved. Prepare Parent Test Class @RunWith(SpringRunner.class)  @SpringBootTest(webEnvironment = WebEnvironment.NONE) @AutoConfigureMessageVerifier  public abstract class MsgTestBase { @Autowired CustomerService service; protected void create() { service.create("@making");   } }

© 2016 Pivotal Software, Inc. All rights reserved. Prepare Parent Test Class @RunWith(SpringRunner.class)  @SpringBootTest(webEnvironment = WebEnvironment.NONE) @AutoConfigureMessageVerifier  public abstract class MsgTestBase { @Autowired CustomerService service; protected void create() { service.create("@making");   } } Created by Producer

© 2016 Pivotal Software, Inc. All rights reserved. Contract DSL Contract.make { label 'create-customer' input { triggeredBy('create()') } outputMessage { sentTo('demo-strm') headers({header('Content-Type':'...')})  body('''{"name":"@making"}''') }} shouldCreateCustomer.groovy

© 2016 Pivotal Software, Inc. All rights reserved. Contract DSL Contract.make { label 'create-customer' input { triggeredBy('create()') } outputMessage { sentTo('demo-strm') headers({header('Content-Type':'...')})  body('''{"name":"@making"}''') }} shouldCreateCustomer.groovy Updated by Producer

© 2016 Pivotal Software, Inc. All rights reserved. Generated Acceptance Test public class ContractVerifierTest extends MsgTestBase { // ... @Test public void validate_shouldCreateCustomer() { create(); ContractVerifierMessage res = verifierMessaging .receive("customer");  assertThat(res).isNotNull(); DocumentContext parsedJson = JsonPath.parse( objectMapper.writeValueAsString(res.getPayload())); assertThatJson(parsedJson).field("name") .isEqualTo("@making");  }}

© 2016 Pivotal Software, Inc. All rights reserved. Spring Cloud Contract Maven Plugin mvn spring-cloud-contract:generateTests mvn spring-cloud-contract:convert mvn spring-cloud-contract:generateStubs Acceptance Test WireMock stub file (only for REST) Stub jar file

© 2016 Pivotal Software, Inc. All rights reserved. Consumer Side Test @RunWith(SpringRunner.class)  @SpringBootTest(webEnvironment = WebEnvironment.NONE) @AutoConfigureStubRunner(ids = "com.example:customerservice", workOffline = true)  public class PointServiceConsumerTest { @Autowired StubTrigger stubTrigger; // ... @Test public void testCreateCustomer() { stubTrigger.trigger("create-customer"); // assert that the message is received ...  } }

© 2016 Pivotal Software, Inc. All rights reserved. (FYI) CQRS and Event Sourcing • https://spring.io/blog/2016/11/08/cqrs-and-event-sourcing- with-jakub-pilimon • https://github.com/pilloPl/event-source-cqrs-sample

© 2016 Pivotal Software, Inc. All rights reserved. Spring Cloud Data Flow • Microservices-based Distributed Data Pipelines •Long Lived Stream Applications •Short Lived Task Applications Spring Cloud Stream

© 2016 Pivotal Software, Inc. All rights reserved. Spring Cloud Data Flow • Microservices-based Distributed Data Pipelines •Long Lived Stream Applications •Short Lived Task Applications Spring Cloud Stream Spring Cloud Task

© 2016 Pivotal Software, Inc. All rights reserved. Spring Cloud Data Flow • Microservices-based Distributed Data Pipelines •Long Lived Stream Applications •Short Lived Task Applications Spring Cloud Stream Spring Cloud Task Orchestration Layer

© 2016 Pivotal Software, Inc. All rights reserved. Cloud Foundry •https://www.cloudfoundry.org/ •Cloud Native Platform •OSS •Spring ❤ Cloud Foundry •Support Multi IaaS   (AWS, Azure, GCP, vSphere, OpenStack)

© 2016 Pivotal Software, Inc. All rights reserved. PCF Dev • https://docs.pivotal.io/pcf-dev • Cloud Foundry on your laptop • Included • Redis / RabbitMQ / MySQL • Spring Cloud Services • Install with cf dev start

© 2016 Pivotal Software, Inc. All rights reserved. Deploy Spring Cloud Stream Apps cf push my-source my-source.jar --no-start cf bind-service my-source my-binder cf start my-source cf push my-sink my-sink.jar --no-start cf bind-service my-sink my-binder cf start my-sink # in case of PCF Dev cf create-service p-rabbitmq standard my-binder # in case of Pivotal Web Services cf create-service cloudamqp lemur my-binder

© 2016 Pivotal Software, Inc. All rights reserved. Sample Application http://bit.ly/making_ccc_a3 Pivotal Web Services / PCF Dev Frontend Customer Service Email Service Point Service Post ServiceZipkin Server MySQL SendGrid MySQL MySQL DLQ Recovery RabbitMQ HTTP

© 2016 Pivotal Software, Inc. All rights reserved. Thanks!! • https://cloud.spring.io/spring-cloud-stream/ • https://cloud.spring.io/spring-cloud-dataflow/ • https://cloud.spring.io/spring-cloud-sleuth/ • http://zipkin.io/ • https://cloud.spring.io/spring-cloud-contract/ • https://projects.spring.io/spring-amqp/ • https://run.pivotal.io/

Event Driven Microservices with Spring Cloud Stream #jjug_ccc #ccc_ab3

More Related Content

What's hot

Viewers also liked

Similar to Event Driven Microservices with Spring Cloud Stream #jjug_ccc #ccc_ab3

More from Toshiaki Maki

Recently uploaded

Event Driven Microservices with Spring Cloud Stream #jjug_ccc #ccc_ab3