noahhsu / event-sourcing-order-poc Goto Github PK

View Code? Open in Web Editor NEW

23.0 4.0 4.0 749 KB

A POC for order status dealing using event sourcing, and add on other cool thing

Java 100.00%

event-sourcing kafka spring-boot

event-sourcing-order-poc's Introduction

Event Sourcing Order P.O.C

Event Stream

Code Structure

Dependency Services

Kafka: Event Store & Event Bus

Domain Aggregate

Order
Payment
Shipment

Server Role

Command-Side:
- handle command request
- write events to Kafka
Query-Side:
- consume events from Kafka to build data view
- return query request
Event-Handler
- consume events from Kafka to send command request to Command-side

How to run application (local code in IDE)

run only dependency component (kafka, redis)

# run all dependency components (kafka, redis, grafana, loki, prometheus, tempo)
docker compose -f Docker/observe-docker-compose.yaml -f Docker/kafka-docker-compose.yml -f Docker/redis-docker-compose.yml -p event-sourcing up

# run kafka and redis only
docker compose -f Docker/kafka-docker-compose.yml -f Docker/redis-docker-compose.yml -p event-sourcing up

# set the working directory as `{path-to-project-root}/event-sourcing-order-poc`
# run the Application you want

How to run application (local code in docker)

With local code

# should run build first
./gradlew build -x test

# run only dependency component (kafka, redis)  
docker compose -f Docker/kafka-docker-compose.yml -f Docker/redis-docker-compose.yml -p event-sourcing up

Run applications

With docker image on the Docker Hub

# all service with all observability components and kafka
docker compose -f Docker/observe-docker-compose.yaml -f Docker/boot-apps-docker-compose.yml -f Docker/kafka-docker-compose.yml -f Docker/redis-docker-compose.yml --profile all -p event-sourcing up
# all order service without observability
docker compose -f Docker/boot-apps-docker-compose.yml -f Docker/kafka-docker-compose.yml -f Docker/redis-docker-compose.yml --env-file Docker/config/.env.docker --profile order -p event-sourcing up --scale order-handler=0
# only order-command-side
docker compose -f Docker/boot-apps-docker-compose.yml -f Docker/kafka-docker-compose.yml -f Docker/redis-docker-compose.yml --env-file Docker/config/.env.docker --profile order -p event-sourcing up --scale order-handler=0 --scale order-query=0
# only payment-command-side
docker compose -f Docker/boot-apps-docker-compose.yml -f Docker/kafka-docker-compose.yml --env-file Docker/config/.env.docker --profile payment -p event-sourcing up --scale payment-handler=0 --scale payment-query=0
# only shipment-command-side
docker compose -f Docker/boot-apps-docker-compose.yml -f Docker/kafka-docker-compose.yml --env-file Docker/config/.env.docker --profile shipment -p event-sourcing up --scale shipment-handler=0 --scale shipment-query=0

Run E2E test in local

prerequisite

install nektos/act (https://github.com/nektos/act)
run docker deamon

command

for mac with M2 chip

act -j e2e-test --container-architecture linux/amd64

for normal linux based kernel

act -j e2e-test

Endpoint

server	swagger url
Order Command	http://localhost:8081/swagger-ui/index.html
Order Handler	http://localhost:8082/swagger-ui/index.html
Order Query	http://localhost:8083/swagger-ui/index.html
Payment Command	http://localhost:9081/swagger-ui/index.html
Payment Handler	http://localhost:9082/swagger-ui/index.html
Payment Query	http://localhost:9083/swagger-ui/index.html
Shipment Command	http://localhost:10081/swagger-ui/index.html
Shipment Handler	http://localhost:10082/swagger-ui/index.html
Shipment Query	http://localhost:10083/swagger-ui/index.html
Grafana	http://localhost:3200
Prometheus	http://localhost:9090

event-sourcing-order-poc's People

Contributors

Stargazers

Watchers

Forkers

negwu01 jhorlamide bassista yogesh2942

event-sourcing-order-poc's Issues

Use KRaft mode for kafka

refer to: https://www.conduktor.io/blog/kraft-quorum-run-kafka-without-zookeeper/

shipment aggregate logic

Task:

Handler: listen Payment Authorized event and send request shipment notify (we can mock one notification or SMS client for now)
Query side: update shipment status for every Shipment event
Command side: send events for every received command

refer to event stream logic

integrate code quality tool

integrate error-prone or other similar tool

Github action push image to docker hub

not for every push on master.
maybe on a release or some special tag,

set the docker access token on the environment variable

API Client module for Order command/query-side

add an API Client module for Order command-side and query-side which is containing

API client for all handlers to use
API Request and Response DTO for order command-side and query-side

this issue should be done before [Issue #10 ]

apply idempotency to command api

refer to https://noahhsu.github.io/System%20Design/How_to_design_an_efficiency_Idempotency_API/

Client Module error handling

try order complete api

get 404 on query side
but return 500 on complete API

2023-04-22T13:09:47.652+08:00 DEBUG [order-command-side,64436c1bb0c954df53174540efe255e8,1129d036f32a7b37] 3086 --- [nio-8081-exec-1] feign.Logger                             : [OrderQueryStub#get] <--- END HTTP (0-byte body)
2023-04-22T13:09:47.655+08:00  INFO [order-command-side,64436c1bb0c954df53174540efe255e8,53174540efe255e8] 3086 --- [nio-8081-exec-1] o.e.e.s.o.p.c.o.c.OrderController        : [OrderController.completeOrder] end
2023-04-22T13:09:47.656+08:00 ERROR [order-command-side,,] 3086 --- [nio-8081-exec-1] o.a.c.c.C.[.[.[/].[dispatcherServlet]    : Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed: feign.FeignException$NotFound: [404] during [GET] to [http://localhost:8083/api/v1-orders/2222] [OrderQueryStub#get(String)]: []] with root cause

feign.FeignException$NotFound: [404] during [GET] to [http://localhost:8083/api/v1-orders/2222] [OrderQueryStub#get(String)]: []
	at feign.FeignException.clientErrorStatus(FeignException.java:228)
	at feign.FeignException.errorStatus(FeignException.java:203)
	at feign.FeignException.errorStatus(FeignException.java:194)
	at feign.FeignException.errorStatus(FeignException.java:171)
	at feign.codec.ErrorDecoder$Default.decode(ErrorDecoder.java:92)
	at feign.ResponseHandler.decodeError(ResponseHandler.java:136)
	at feign.ResponseHandler.handleResponse(ResponseHandler.java:70)
	at feign.SynchronousMethodHandler.executeAndDecode(SynchronousMethodHandler.java:116)
	at feign.SynchronousMethodHandler.invoke(SynchronousMethodHandler.java:72)
	at feign.ReflectiveFeign$FeignInvocationHandler.invoke(ReflectiveFeign.java:98)
	at feign.micrometer.MeteredInvocationHandleFactory.lambda$create$0(MeteredInvocationHandleFactory.java:70)
	at org.example.event.sourcing.order.poc.client.order.$Proxy100.get(Unknown Source)
	at org.example.event.sourcing.order.poc.client.order.OrderQueryClient.get(OrderQueryClient.java:20)
	at org.example.event.sourcing.order.poc.command.order.service.OrderService.completeOrder(OrderService.java:40)
	at org.example.event.sourcing.order.poc.command.order.controller.OrderController.completeOrder(OrderController.java:30)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104)
	at java.base/java.lang.reflect.Method.invoke(Method.java:578)
	at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:343)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:196)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
	at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:752)
	at org.springframework.aop.aspectj.MethodInvocationProceedingJoinPoint.proceed(MethodInvocationProceedingJoinPoint.java:89)
	at io.micrometer.observation.aop.ObservedAspect.lambda$observe$2(ObservedAspect.java:150)
	at io.micrometer.observation.Observation.observeChecked(Observation.java:594)
	at io.micrometer.observation.aop.ObservedAspect.observe(ObservedAspect.java:150)
	at io.micrometer.observation.aop.ObservedAspect.observeClass(ObservedAspect.java:116)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104)
	at java.base/java.lang.reflect.Method.invoke(Method.java:578)
	at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethodWithGivenArgs(AbstractAspectJAdvice.java:633)
	at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethod(AbstractAspectJAdvice.java:623)
	at org.springframework.aop.aspectj.AspectJAroundAdvice.invoke(AspectJAroundAdvice.java:71)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:173)
	at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:752)
	at org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:97)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:184)
	at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:752)
	at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:703)
	at org.example.event.sourcing.order.poc.command.order.controller.OrderController$$SpringCGLIB$$0.completeOrder(<generated>)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104)
	at java.base/java.lang.reflect.Method.invoke(Method.java:578)
	at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:207)
	at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:152)
	at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:117)
	at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:884)
	at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:797)
	at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:87)
	at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:1080)
	at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:973)
	at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:1003)
	at org.springframework.web.servlet.FrameworkServlet.doPost(FrameworkServlet.java:906)
	at jakarta.servlet.http.HttpServlet.service(HttpServlet.java:731)
	at org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:880)
	at jakarta.servlet.http.HttpServlet.service(HttpServlet.java:814)
	at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:223)
	at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:158)
	at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:53)
	at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:185)
	at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:158)
	at org.springframework.web.filter.RequestContextFilter.doFilterInternal(RequestContextFilter.java:100)
	at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:116)
	at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:185)
	at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:158)
	at org.springframework.web.filter.FormContentFilter.doFilterInternal(FormContentFilter.java:93)
	at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:116)
	at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:185)
	at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:158)
	at org.springframework.web.filter.ServerHttpObservationFilter.doFilterInternal(ServerHttpObservationFilter.java:109)
	at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:116)
	at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:185)
	at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:158)
	at org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:201)
	at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:116)
	at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:185)
	at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:158)
	at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:197)
	at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:97)
	at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:542)
	at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:119)
	at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:92)
	at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:78)
	at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:357)
	at org.apache.coyote.http11.Http11Processor.service(Http11Processor.java:400)
	at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:65)
	at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:861)
	at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1739)
	at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:52)
	at org.apache.tomcat.util.threads.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1191)
	at org.apache.tomcat.util.threads.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:659)
	at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)
	at java.base/java.lang.Thread.run(Thread.java:1589)

As-is:

unknown
find out the partition setting for now (bitnami/kafka default value)

To-be:

partition divided by the id of each aggregate
find a proper # of partition

Move to Spring boot 3.0

https://github.com/spring-projects/spring-boot/wiki/Spring-Boot-3.0-Migration-Guide

refine document

update README for run application guide
add pr template

Check whether exist before send created event for order aggregate

For now, create order command always sends create an event to the Kafka,
which might cause duplication on the order query side when handling the event.

need to prevent this issue by checking if the order in the request body was created already.

API Client module for Shipment command-side

add an API Client module for Shipment command-side which is containing

API client for all handlers to use
API Request and Response DTO for shipment command-side and all handlers

support backstage

https://backstage.io/docs/overview/what-is-backstage

update ignore file for Eclipse

please have mercy for Eclipse users

project observability

provide health check, Prometheus...etc monitoring api

reference article: https://spring.io/blog/2022/10/12/observability-with-spring-boot-3

payment aggregate logic

Should start after [Issue #29]

Task:

Handler:
- listen Order Created event then send request payment notify (we can mock one notification or SMS client for now)
- listen Payment Created event then call valid payment command
- listen Payment Unauthorized event then call cancel payment command
- listen Shipment Received event then call settle payment command
Query side: update payment status for every Payment event
Command side: send events for every received command

refer to event stream logic

Discussion: this is off topic but I personally don't like to specify packages to scan. Maybe we can borrow from spring-boot's pattern and find a cleaner way to do this, for example custom EnableXXX annotation ?

Originally posted by @woodyf in #39 (comment)

kafka error handling mechanism

Maybe use retry topic and fail topic.
Find if there is an library already implements it

build native image with Graalvm

integrate with Jib
offial extension

order aggregate logic

Should start after [Issue #29]

Task:

Handler: listen Payment Settled event then call complete order command
design the package scan for scan API client bean ref
Query side: update order status for every Order event
Command side: send events for every received command

refer to event stream logic

Let both applications run in docker compose and on host machine access kafka (ip/hostname problem)

~~docker network mode = host~~ Only avaliable in linux but not mac

use host.docker.internal, and map host.docker.internal to 127.0.0.1 on host machine via editing hosts file
-> this is the best we've got by far.

Is there a better way?

Refactor code Structure

API Client module for Payment command-side

add an API Client module for Payment command-side which is containing

API client for all handlers to use
API Request and Response DTO for payment command-side and all handlers

this issue should be done before [Issue #11]

performance
reproducible
- GoogleContainerTools/jib#2143

reference:

multi nondule example

merge all the Swagger UI into a single one

It is not in this PR scope, but I think we can merge all the Swagger UI into a single one

Originally posted by @janbarry168 in #49 (comment)

Recommend Projects

React

A declarative, efficient, and flexible JavaScript library for building user interfaces.
Vue.js

🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
Typescript

TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
TensorFlow

An Open Source Machine Learning Framework for Everyone
Django

The Web framework for perfectionists with deadlines.
Laravel

A PHP framework for web artisans
D3

Bring data to life with SVG, Canvas and HTML. 📊📈🎉

Recommend Topics

javascript

JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
web

Some thing interesting about web. New door for the world.
server

A server is a program made to process requests and deliver data to clients.
Machine learning

Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Visualization

Some thing interesting about visualization, use data art
Game

Some thing interesting about game, make everyone happy.

Recommend Org

Facebook

We are working to build community through open source technology. NB: members must have two-factor auth.
Microsoft

Open source projects and samples from Microsoft.
Google

Google ❤️ Open Source for everyone.
Alibaba

Alibaba Open Source for everyone
D3

Data-Driven Documents codes.
Tencent

China tencent open source team.