3 Commits

Author SHA1 Message Date
Alexander
5275894580 feature: add jaeger opentracing improvements 2022-04-14 15:30:08 +03:00
Alexander
ea25e8e8bc feature: add jaeger opentracing 2022-04-14 15:18:33 +03:00
Alexander
501b5a9cec feature: add prometheus monitoring 2022-04-14 15:04:44 +03:00
30 changed files with 200 additions and 4858 deletions

3
.gitignore vendored
View File

@@ -35,5 +35,4 @@ kafka_data
zookeeper
pgdata
prometheus
mongodb_data_container
grafana
mongodb_data_container

View File

@@ -1,5 +1,27 @@
.PHONY:
#Quarkus
## Running the application in dev mode
#You can run your application in dev mode that enables live coding using:
dev:
./mvnw compile quarkus:dev
# The application can be packaged using:
package:
./mvnw package
package-uber-jar:
./mvnw package -Dquarkus.package.type=uber-jar
package-native:
./mvnw package -Pnative
package-native-GraalVM:
./mvnw package -Pnative -Dquarkus.native.container-build=true
# ==============================================================================
# Docker

View File

@@ -1,39 +0,0 @@
### Spring CQRS and Event Sourcing microservice example 👋‍💫✨
#### 👨‍💻 Full list what has been used:
* [Spring](https://spring.io/) - Java Spring
* [Spring Data JPA](https://spring.io/projects/spring-data-jpa) - data access layer
* [Spring Data MongoDB](https://spring.io/projects/spring-data-mongodb) - Spring Data MongoDB
* [Spring Cloud Sleuth](https://spring.io/projects/spring-cloud-sleuth) - Spring Cloud Sleuth distributed tracing
* [Kafka](https://spring.io/projects/spring-kafka) - Spring for Apache Kafka
* [PostgreSQL](https://www.postgresql.org/) - PostgreSQL database.
* [Jaeger](https://www.jaegertracing.io/) - Jaeger is a distributed tracing system
* [Docker](https://www.docker.com/) - Docker
* [Prometheus](https://prometheus.io/) - Prometheus
* [Grafana](https://grafana.com/) - Grafana
* [Flyway](https://flywaydb.org/) - Database migrations
* [Resilience4j](https://resilience4j.readme.io/docs/getting-started-3) - Resilience4j is a lightweight, easy-to-use fault tolerance
* [Swagger OpenAPI 3](https://springdoc.org/) - java library helps to automate the generation of API documentation
### Swagger UI:
http://localhost:8006/swagger-ui/index.html
### Jaeger UI:
http://localhost:16686
### Prometheus UI:
http://localhost:9090
### Grafana UI:
http://localhost:3005
For local development:
```
make local // runs docker-compose.yaml with all required containers
run spring application
```

View File

@@ -102,9 +102,7 @@ services:
grafana:
container_name: grafana_container
restart: always
build: monitoring/grafana
volumes:
- ./grafana:/var/lib/grafana
image: grafana/grafana
ports:
- '3005:3000'
networks: [ "microservices" ]

View File

@@ -1,2 +0,0 @@
FROM grafana/grafana
ADD ./provisioning /etc/grafana/provisioning

View File

@@ -1,12 +0,0 @@
# grafana/provisioning/dashboards/all.yaml
apiVersion: 1
providers:
- name: 'default'
folder: 'default'
type: file
allowUiUpdates: true
updateIntervalSeconds: 30
options:
path: /etc/grafana/provisioning/dashboards
foldersFromFilesStructure: true

View File

@@ -1,816 +0,0 @@
{
"annotations": {
"list": [
{
"builtIn": 1,
"datasource": "-- Grafana --",
"enable": true,
"hide": true,
"iconColor": "rgba(0, 211, 255, 1)",
"name": "Annotations & Alerts",
"type": "dashboard"
}
]
},
"description": "Micrometer Http throughput dashboard ",
"editable": true,
"gnetId": 5373,
"graphTooltip": 0,
"id": 5,
"iteration": 1615648797248,
"links": [],
"panels": [
{
"collapsed": false,
"datasource": null,
"gridPos": {
"h": 1,
"w": 24,
"x": 0,
"y": 0
},
"id": 2,
"panels": [],
"repeat": null,
"title": "Status and SLA",
"type": "row"
},
{
"cacheTimeout": null,
"colorBackground": false,
"colorValue": false,
"colors": [
"#299c46",
"rgba(237, 129, 40, 0.89)",
"#d44a3a"
],
"datasource": "Prometheus",
"fieldConfig": {
"defaults": {
"custom": {}
},
"overrides": []
},
"format": "none",
"gauge": {
"maxValue": 100,
"minValue": 0,
"show": false,
"thresholdLabels": false,
"thresholdMarkers": true
},
"gridPos": {
"h": 4,
"w": 2,
"x": 0,
"y": 1
},
"id": 4,
"interval": null,
"links": [],
"mappingType": 1,
"mappingTypes": [
{
"name": "value to text",
"value": 1
},
{
"name": "range to text",
"value": 2
}
],
"maxDataPoints": 100,
"nullPointMode": "connected",
"nullText": null,
"postfix": "",
"postfixFontSize": "50%",
"prefix": "",
"prefixFontSize": "50%",
"rangeMaps": [
{
"from": "null",
"text": "N/A",
"to": "null"
}
],
"sparkline": {
"fillColor": "rgba(31, 118, 189, 0.18)",
"full": false,
"lineColor": "rgb(31, 120, 193)",
"show": false
},
"tableColumn": "",
"targets": [
{
"expr": "up{application=\"$application\",instance=\"$instance\"}",
"format": "time_series",
"intervalFactor": 1,
"refId": "A"
}
],
"thresholds": "",
"title": "Status",
"type": "singlestat",
"valueFontSize": "80%",
"valueMaps": [
{
"op": "=",
"text": "UP",
"value": "1"
},
{
"op": "=",
"text": "DOWN",
"value": "0"
},
{
"op": "=",
"text": "DOWN",
"value": "null"
}
],
"valueName": "avg"
},
{
"columns": [],
"datasource": "Prometheus",
"fieldConfig": {
"defaults": {
"custom": {}
},
"overrides": []
},
"fontSize": "100%",
"gridPos": {
"h": 8,
"w": 22,
"x": 2,
"y": 1
},
"hideTimeOverride": false,
"id": 26,
"links": [],
"pageSize": null,
"scroll": true,
"showHeader": true,
"sort": {
"col": 0,
"desc": true
},
"styles": [
{
"alias": "Time",
"align": "auto",
"dateFormat": "YYYY-MM-DD HH:mm:ss",
"pattern": "Time",
"type": "date"
},
{
"alias": "",
"align": "auto",
"colorMode": null,
"colors": [
"rgba(245, 54, 54, 0.9)",
"rgba(237, 129, 40, 0.89)",
"rgba(50, 172, 45, 0.97)"
],
"decimals": 2,
"pattern": "/.*/",
"thresholds": [],
"type": "number",
"unit": "short"
}
],
"targets": [
{
"expr": "http_server_requests_seconds_bucket{application=\"$application\", instance=\"$instance\", le=\"0.05\"}",
"format": "table",
"instant": true,
"intervalFactor": 1,
"legendFormat": "",
"refId": "A"
}
],
"title": "Number Requests that missed 50ms SLA",
"transform": "table",
"type": "table-old"
},
{
"cacheTimeout": null,
"colorBackground": false,
"colorValue": true,
"colors": [
"#299c46",
"rgba(237, 129, 40, 0.89)",
"#d44a3a"
],
"datasource": "Prometheus",
"decimals": 1,
"fieldConfig": {
"defaults": {
"custom": {}
},
"overrides": []
},
"format": "s",
"gauge": {
"maxValue": 100,
"minValue": 0,
"show": false,
"thresholdLabels": false,
"thresholdMarkers": true
},
"gridPos": {
"h": 4,
"w": 2,
"x": 0,
"y": 5
},
"id": 18,
"interval": null,
"links": [],
"mappingType": 1,
"mappingTypes": [
{
"name": "value to text",
"value": 1
},
{
"name": "range to text",
"value": 2
}
],
"maxDataPoints": 100,
"nullPointMode": "connected",
"nullText": null,
"postfix": "",
"postfixFontSize": "50%",
"prefix": "",
"prefixFontSize": "50%",
"rangeMaps": [
{
"from": "null",
"text": "N/A",
"to": "null"
}
],
"sparkline": {
"fillColor": "rgba(31, 118, 189, 0.18)",
"full": false,
"lineColor": "rgb(31, 120, 193)",
"show": false
},
"tableColumn": "",
"targets": [
{
"expr": "process_uptime_seconds{application=\"$application\", instance=\"$instance\"}",
"format": "time_series",
"intervalFactor": 1,
"refId": "A"
}
],
"thresholds": "",
"title": "Up Time",
"type": "singlestat",
"valueFontSize": "80%",
"valueMaps": [
{
"op": "=",
"text": "N/A",
"value": "null"
}
],
"valueName": "current"
},
{
"collapsed": false,
"datasource": null,
"gridPos": {
"h": 1,
"w": 24,
"x": 0,
"y": 9
},
"id": 8,
"panels": [],
"title": "Throughput",
"type": "row"
},
{
"aliasColors": {},
"bars": false,
"dashLength": 10,
"dashes": false,
"datasource": "Prometheus",
"fieldConfig": {
"defaults": {
"custom": {}
},
"overrides": []
},
"fill": 1,
"fillGradient": 0,
"gridPos": {
"h": 9,
"w": 12,
"x": 0,
"y": 10
},
"hiddenSeries": false,
"id": 10,
"legend": {
"avg": false,
"current": true,
"max": true,
"min": false,
"show": true,
"total": false,
"values": true
},
"lines": true,
"linewidth": 2,
"links": [],
"nullPointMode": "null",
"options": {
"alertThreshold": true
},
"percentage": false,
"pluginVersion": "7.4.3",
"pointradius": 5,
"points": false,
"renderer": "flot",
"seriesOverrides": [],
"spaceLength": 10,
"stack": false,
"steppedLine": false,
"targets": [
{
"expr": "rate(http_server_requests_seconds_count{application=\"$application\", instance=\"$instance\"}[1m])",
"format": "time_series",
"intervalFactor": 1,
"legendFormat": "{{method}}-{{status}}-{{uri}}",
"refId": "A"
}
],
"thresholds": [],
"timeFrom": null,
"timeRegions": [],
"timeShift": null,
"title": "Requests per second",
"tooltip": {
"shared": true,
"sort": 0,
"value_type": "individual"
},
"type": "graph",
"xaxis": {
"buckets": null,
"mode": "time",
"name": null,
"show": true,
"values": []
},
"yaxes": [
{
"format": "short",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
},
{
"format": "short",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
}
],
"yaxis": {
"align": false,
"alignLevel": null
}
},
{
"aliasColors": {},
"bars": false,
"dashLength": 10,
"dashes": false,
"datasource": "Prometheus",
"fieldConfig": {
"defaults": {
"custom": {}
},
"overrides": []
},
"fill": 1,
"fillGradient": 0,
"gridPos": {
"h": 9,
"w": 12,
"x": 12,
"y": 10
},
"hiddenSeries": false,
"id": 12,
"legend": {
"avg": false,
"current": false,
"hideEmpty": false,
"hideZero": false,
"max": false,
"min": false,
"show": true,
"total": false,
"values": false
},
"lines": true,
"linewidth": 2,
"links": [],
"nullPointMode": "null as zero",
"options": {
"alertThreshold": true
},
"percentage": false,
"pluginVersion": "7.4.3",
"pointradius": 5,
"points": false,
"renderer": "flot",
"seriesOverrides": [],
"spaceLength": 10,
"stack": false,
"steppedLine": false,
"targets": [
{
"expr": "rate(http_server_requests_seconds_sum{application=\"$application\", instance=\"$instance\"}[1m])/rate(http_server_requests_seconds_count{application=\"$application\", instance=\"$instance\"}[1m])",
"format": "time_series",
"instant": false,
"intervalFactor": 1,
"legendFormat": "{{method}}-{{status}}-{{uri}}",
"refId": "A"
}
],
"thresholds": [],
"timeFrom": null,
"timeRegions": [],
"timeShift": null,
"title": "Mean response time",
"tooltip": {
"shared": true,
"sort": 0,
"value_type": "individual"
},
"type": "graph",
"xaxis": {
"buckets": null,
"mode": "time",
"name": null,
"show": true,
"values": []
},
"yaxes": [
{
"format": "s",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
},
{
"format": "short",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
}
],
"yaxis": {
"align": false,
"alignLevel": null
}
},
{
"aliasColors": {},
"bars": false,
"dashLength": 10,
"dashes": false,
"datasource": "Prometheus",
"fieldConfig": {
"defaults": {
"custom": {}
},
"overrides": []
},
"fill": 1,
"fillGradient": 0,
"gridPos": {
"h": 9,
"w": 12,
"x": 0,
"y": 19
},
"hiddenSeries": false,
"id": 14,
"legend": {
"avg": false,
"current": true,
"max": false,
"min": false,
"show": true,
"total": false,
"values": true
},
"lines": true,
"linewidth": 2,
"links": [],
"nullPointMode": "null",
"options": {
"alertThreshold": true
},
"percentage": false,
"pluginVersion": "7.4.3",
"pointradius": 5,
"points": false,
"renderer": "flot",
"seriesOverrides": [],
"spaceLength": 10,
"stack": false,
"steppedLine": false,
"targets": [
{
"expr": "histogram_quantile(0.95, sum(rate(http_server_requests_seconds_bucket{application=\"$application\", instance=\"$instance\"}[1m])) by (le))",
"format": "time_series",
"instant": false,
"intervalFactor": 1,
"legendFormat": "95%",
"refId": "A"
},
{
"expr": "histogram_quantile(0.9, sum(rate(http_server_requests_seconds_bucket{application=\"$application\", instance=\"$instance\"}[1m])) by (le))",
"format": "time_series",
"intervalFactor": 1,
"legendFormat": "90%",
"refId": "B"
},
{
"expr": "histogram_quantile(0.75, sum(rate(http_server_requests_seconds_bucket{application=\"$application\", instance=\"$instance\"}[1m])) by (le))",
"format": "time_series",
"intervalFactor": 1,
"legendFormat": "75%",
"refId": "C"
},
{
"expr": "histogram_quantile(0.5, sum(rate(http_server_requests_seconds_bucket{application=\"$application\", instance=\"$instance\"}[1m])) by (le))",
"format": "time_series",
"intervalFactor": 1,
"legendFormat": "50%",
"refId": "D"
}
],
"thresholds": [],
"timeFrom": null,
"timeRegions": [],
"timeShift": null,
"title": "Response time of 50%, 75%, 90%, 95% of requests",
"tooltip": {
"shared": true,
"sort": 0,
"value_type": "individual"
},
"type": "graph",
"xaxis": {
"buckets": null,
"mode": "time",
"name": null,
"show": true,
"values": []
},
"yaxes": [
{
"format": "s",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
},
{
"format": "short",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
}
],
"yaxis": {
"align": false,
"alignLevel": null
}
},
{
"aliasColors": {},
"bars": false,
"dashLength": 10,
"dashes": false,
"datasource": "Prometheus",
"fieldConfig": {
"defaults": {
"custom": {}
},
"overrides": []
},
"fill": 1,
"fillGradient": 0,
"gridPos": {
"h": 9,
"w": 12,
"x": 12,
"y": 19
},
"hiddenSeries": false,
"id": 16,
"legend": {
"avg": false,
"current": false,
"max": false,
"min": false,
"show": true,
"total": false,
"values": false
},
"lines": true,
"linewidth": 1,
"links": [],
"nullPointMode": "null as zero",
"options": {
"alertThreshold": true
},
"percentage": false,
"pluginVersion": "7.4.3",
"pointradius": 5,
"points": false,
"renderer": "flot",
"seriesOverrides": [],
"spaceLength": 10,
"stack": false,
"steppedLine": false,
"targets": [
{
"expr": "topk(10, sum by(uri, method) (rate(http_server_requests_seconds_count{application=\"$application\"}[1m])))",
"format": "time_series",
"intervalFactor": 1,
"legendFormat": "",
"refId": "A"
}
],
"thresholds": [],
"timeFrom": null,
"timeRegions": [],
"timeShift": null,
"title": "Top 10 APIs",
"tooltip": {
"shared": true,
"sort": 0,
"value_type": "individual"
},
"type": "graph",
"xaxis": {
"buckets": null,
"mode": "time",
"name": null,
"show": true,
"values": []
},
"yaxes": [
{
"format": "short",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
},
{
"format": "short",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
}
],
"yaxis": {
"align": false,
"alignLevel": null
}
}
],
"schemaVersion": 27,
"style": "dark",
"tags": [],
"templating": {
"list": [
{
"allValue": null,
"current": {
"isNone": true,
"selected": false,
"text": "None",
"value": ""
},
"datasource": "Prometheus",
"definition": "",
"description": null,
"error": null,
"hide": 0,
"includeAll": false,
"label": "Application",
"multi": false,
"name": "application",
"options": [],
"query": {
"query": "label_values(application)",
"refId": "Prometheus-application-Variable-Query"
},
"refresh": 2,
"regex": "",
"skipUrlSync": false,
"sort": 0,
"tagValuesQuery": "",
"tags": [],
"tagsQuery": "",
"type": "query",
"useTags": false
},
{
"allValue": null,
"current": {
"selected": false,
"text": "host.docker.internal:8080",
"value": "host.docker.internal:8080"
},
"datasource": "Prometheus",
"definition": "",
"description": null,
"error": null,
"hide": 0,
"includeAll": false,
"label": "instance",
"multi": false,
"name": "instance",
"options": [],
"query": {
"query": "label_values(up{application=\"$application\"}, instance)",
"refId": "Prometheus-instance-Variable-Query"
},
"refresh": 2,
"regex": "",
"skipUrlSync": false,
"sort": 0,
"tagValuesQuery": "",
"tags": [],
"tagsQuery": "",
"type": "query",
"useTags": false
}
]
},
"time": {
"from": "now-15m",
"to": "now"
},
"timepicker": {
"refresh_intervals": [
"5s",
"10s",
"30s",
"1m",
"5m",
"15m",
"30m",
"1h",
"2h",
"1d"
],
"time_options": [
"5m",
"15m",
"1h",
"6h",
"12h",
"24h",
"2d",
"7d",
"30d"
]
},
"timezone": "",
"title": "Micrometer Spring Throughput",
"uid": "twqdYjziz",
"version": 1
}

View File

@@ -1,9 +0,0 @@
apiVersion: 1
datasources:
- name: Prometheus
label: Prometheus
type: prometheus
access: proxy
url: http://localhost:9090
isDefault: true

View File

@@ -17,11 +17,6 @@
<java.version>17</java.version>
</properties>
<dependencies>
<dependency>
<groupId>io.github.resilience4j</groupId>
<artifactId>resilience4j-spring-boot2</artifactId>
<version>1.7.1</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-sleuth-zipkin</artifactId>

View File

@@ -3,8 +3,6 @@ package com.eventsourcing.bankAccount.commands;
import com.eventsourcing.bankAccount.domain.BankAccountAggregate;
import com.eventsourcing.es.EventStoreDB;
import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import io.github.resilience4j.retry.annotation.Retry;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.sleuth.annotation.NewSpan;
@@ -17,12 +15,9 @@ import org.springframework.stereotype.Service;
public class BankAccountCommandHandler implements BankAccountCommandService {
private final EventStoreDB eventStoreDB;
private static final String SERVICE_NAME = "microservice";
@Override
@NewSpan
@Retry(name = SERVICE_NAME)
@CircuitBreaker(name = SERVICE_NAME)
public String handle(@SpanTag("command") CreateBankAccountCommand command) {
final var aggregate = new BankAccountAggregate(command.aggregateID());
aggregate.createBankAccount(command.email(), command.address(), command.userName());
@@ -34,8 +29,6 @@ public class BankAccountCommandHandler implements BankAccountCommandService {
@Override
@NewSpan
@Retry(name = SERVICE_NAME)
@CircuitBreaker(name = SERVICE_NAME)
public void handle(@SpanTag("command") ChangeEmailCommand command) {
final var aggregate = eventStoreDB.load(command.aggregateID(), BankAccountAggregate.class);
aggregate.changeEmail(command.newEmail());
@@ -45,8 +38,6 @@ public class BankAccountCommandHandler implements BankAccountCommandService {
@Override
@NewSpan
@Retry(name = SERVICE_NAME)
@CircuitBreaker(name = SERVICE_NAME)
public void handle(@SpanTag("command") ChangeAddressCommand command) {
final var aggregate = eventStoreDB.load(command.aggregateID(), BankAccountAggregate.class);
aggregate.changeAddress(command.newAddress());
@@ -56,8 +47,6 @@ public class BankAccountCommandHandler implements BankAccountCommandService {
@Override
@NewSpan
@Retry(name = SERVICE_NAME)
@CircuitBreaker(name = SERVICE_NAME)
public void handle(@SpanTag("command") DepositAmountCommand command) {
final var aggregate = eventStoreDB.load(command.aggregateID(), BankAccountAggregate.class);
aggregate.depositBalance(command.amount());

View File

@@ -4,11 +4,9 @@ package com.eventsourcing.bankAccount.delivery;
import com.eventsourcing.bankAccount.commands.*;
import com.eventsourcing.bankAccount.dto.*;
import com.eventsourcing.bankAccount.queries.BankAccountQueryService;
import com.eventsourcing.bankAccount.queries.FindAllOrderByBalance;
import com.eventsourcing.bankAccount.queries.GetBankAccountByIDQuery;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.domain.Page;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
@@ -27,43 +25,43 @@ public class BankAccountController {
@GetMapping("{aggregateId}")
public ResponseEntity<BankAccountResponseDTO> getBankAccount(@PathVariable String aggregateId) {
final var result = queryService.handle(new GetBankAccountByIDQuery(aggregateId));
log.info("Get bank account result: {}", result);
final var query = new GetBankAccountByIDQuery(aggregateId);
log.info("GET bank account query: {}", query);
final var result = queryService.handle(query);
log.info("GET bank account result: {}", result);
return ResponseEntity.ok(result);
}
@PostMapping
public ResponseEntity<String> createBankAccount(@Valid @RequestBody CreateBankAccountRequestDTO dto) {
final var aggregateID = UUID.randomUUID().toString();
final var id = commandService.handle(new CreateBankAccountCommand(aggregateID, dto.email(), dto.userName(), dto.address()));
log.info("Created bank account id: {}", id);
final var command = new CreateBankAccountCommand(aggregateID, dto.email(), dto.userName(), dto.address());
final var id = commandService.handle(command);
log.info("CREATE bank account id: {}", id);
return ResponseEntity.status(HttpStatus.CREATED).body(id);
}
@PostMapping(path = "/deposit/{aggregateId}")
public ResponseEntity<Void> depositAmount(@Valid @RequestBody DepositAmountRequestDTO dto, @PathVariable String aggregateId) {
commandService.handle(new DepositAmountCommand(aggregateId, dto.amount()));
final var command = new DepositAmountCommand(aggregateId, dto.amount());
commandService.handle(command);
log.info("DepositAmountCommand command: {}", command);
return ResponseEntity.ok().build();
}
@PostMapping(path = "/email/{aggregateId}")
public ResponseEntity<Void> changeEmail(@Valid @RequestBody ChangeEmailRequestDTO dto, @PathVariable String aggregateId) {
commandService.handle(new ChangeEmailCommand(aggregateId, dto.newEmail()));
final var command = new ChangeEmailCommand(aggregateId, dto.newEmail());
commandService.handle(command);
log.info("ChangeEmailCommand command: {}", command);
return ResponseEntity.ok().build();
}
@PostMapping(path = "/address/{aggregateId}")
public ResponseEntity<Void> changeAddress(@Valid @RequestBody ChangeAddressRequestDTO dto, @PathVariable String aggregateId) {
commandService.handle(new ChangeAddressCommand(aggregateId, dto.newAddress()));
final var command = new ChangeAddressCommand(aggregateId, dto.newAddress());
commandService.handle(command);
log.info("changeAddress command: {}", command);
return ResponseEntity.ok().build();
}
@GetMapping("/balance")
public ResponseEntity<Page<BankAccountResponseDTO>> getAllOrderByBalance(@RequestParam(name = "page", defaultValue = "0") Integer page,
@RequestParam(name = "size", defaultValue = "10") Integer size) {
final var result = queryService.handle(new FindAllOrderByBalance(page, size));
log.info("Get all by balance page: {}, size: {}, result: {}", page, size, result);
return ResponseEntity.ok(result);
}
}

View File

@@ -4,8 +4,6 @@ import com.eventsourcing.bankAccount.events.AddressUpdatedEvent;
import com.eventsourcing.bankAccount.events.BalanceDepositedEvent;
import com.eventsourcing.bankAccount.events.BankAccountCreatedEvent;
import com.eventsourcing.bankAccount.events.EmailChangedEvent;
import com.eventsourcing.bankAccount.exceptions.InvalidAddressException;
import com.eventsourcing.bankAccount.exceptions.InvalidEmailException;
import com.eventsourcing.es.AggregateRoot;
import com.eventsourcing.es.Event;
import com.eventsourcing.es.SerializerUtils;
@@ -59,13 +57,13 @@ public class BankAccountAggregate extends AggregateRoot {
private void handle(final EmailChangedEvent event) {
Objects.requireNonNull(event.getNewEmail());
if (event.getNewEmail().isBlank()) throw new InvalidEmailException();
if (event.getNewEmail().isBlank()) throw new RuntimeException("invalid email address");
this.email = event.getNewEmail();
}
private void handle(final AddressUpdatedEvent event) {
Objects.requireNonNull(event.getNewAddress());
if (event.getNewAddress().isBlank()) throw new InvalidAddressException();
if (event.getNewAddress().isBlank()) throw new RuntimeException("invalid address");
this.address = event.getNewAddress();
}

View File

@@ -1,10 +0,0 @@
package com.eventsourcing.bankAccount.exceptions;
public class BankAccountDocumentNotFoundException extends RuntimeException {
public BankAccountDocumentNotFoundException() {
}
public BankAccountDocumentNotFoundException(String id) {
super("bank account document not found id:" + id);
}
}

View File

@@ -1,11 +0,0 @@
package com.eventsourcing.bankAccount.exceptions;
public class InvalidAddressException extends RuntimeException {
public InvalidAddressException() {
super();
}
public InvalidAddressException(String address) {
super("invalid address: " + address);
}
}

View File

@@ -1,11 +0,0 @@
package com.eventsourcing.bankAccount.exceptions;
public class InvalidEmailException extends RuntimeException {
public InvalidEmailException() {
super();
}
public InvalidEmailException(String email) {
super("invalid email address: " + email);
}
}

View File

@@ -7,15 +7,12 @@ import com.eventsourcing.bankAccount.events.AddressUpdatedEvent;
import com.eventsourcing.bankAccount.events.BalanceDepositedEvent;
import com.eventsourcing.bankAccount.events.BankAccountCreatedEvent;
import com.eventsourcing.bankAccount.events.EmailChangedEvent;
import com.eventsourcing.bankAccount.exceptions.BankAccountDocumentNotFoundException;
import com.eventsourcing.bankAccount.repository.BankAccountMongoRepository;
import com.eventsourcing.es.Event;
import com.eventsourcing.es.EventStoreDB;
import com.eventsourcing.es.Projection;
import com.eventsourcing.es.SerializerUtils;
import com.eventsourcing.mappers.BankAccountMapper;
import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import io.github.resilience4j.retry.annotation.Retry;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.sleuth.annotation.NewSpan;
@@ -29,6 +26,7 @@ import org.springframework.stereotype.Service;
import java.math.BigDecimal;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
@Service
@Slf4j
@@ -37,23 +35,24 @@ public class BankAccountMongoProjection implements Projection {
private final BankAccountMongoRepository mongoRepository;
private final EventStoreDB eventStoreDB;
private static final String SERVICE_NAME = "microservice";
@KafkaListener(topics = {"${microservice.kafka.topics.bank-account-event-store}"},
groupId = "${microservice.kafka.groupId}",
concurrency = "${microservice.kafka.default-concurrency}")
public void bankAccountMongoProjectionListener(@Payload byte[] data, ConsumerRecordMetadata meta, Acknowledgment ack) {
log.info("(BankAccountMongoProjection) topic: {}, offset: {}, partition: {}, timestamp: {}, data: {}", meta.topic(), meta.offset(), meta.partition(), meta.timestamp(), new String(data));
log.info("(BankAccountMongoProjection) topic: {}, offset: {}, partition: {}, timestamp: {}", meta.topic(), meta.offset(), meta.partition(), meta.timestamp());
log.info("(BankAccountMongoProjection) data: {}", new String(data));
try {
final Event[] events = SerializerUtils.deserializeEventsFromJsonBytes(data);
this.processEvents(Arrays.stream(events).toList());
ack.acknowledge();
log.info("ack events: {}", Arrays.toString(events));
} catch (Exception ex) {
} catch (Exception e) {
ack.nack(100);
log.error("(BankAccountMongoProjection) topic: {}, offset: {}, partition: {}, timestamp: {}", meta.topic(), meta.offset(), meta.partition(), meta.timestamp(), ex);
log.error("(BankAccountMongoProjection) topic: {}, offset: {}, partition: {}, timestamp: {}", meta.topic(), meta.offset(), meta.partition(), meta.timestamp());
log.error("bankAccountMongoProjectionListener: {}", e.getMessage());
}
}
@@ -74,8 +73,6 @@ public class BankAccountMongoProjection implements Projection {
@Override
@NewSpan
@Retry(name = SERVICE_NAME)
@CircuitBreaker(name = SERVICE_NAME)
public void when(@SpanTag("event") Event event) {
final var aggregateId = event.getAggregateId();
log.info("(when) >>>>> aggregateId: {}", aggregateId);
@@ -113,9 +110,9 @@ public class BankAccountMongoProjection implements Projection {
@NewSpan
private void handle(@SpanTag("event") EmailChangedEvent event) {
log.info("(when) EmailChangedEvent: {}, aggregateID: {}", event, event.getAggregateId());
final var documentOptional = mongoRepository.findByAggregateId(event.getAggregateId());
Optional<BankAccountDocument> documentOptional = mongoRepository.findByAggregateId(event.getAggregateId());
if (documentOptional.isEmpty())
throw new BankAccountDocumentNotFoundException(event.getAggregateId());
throw new RuntimeException("Bank Account Document not found id: {}" + event.getAggregateId());
final var document = documentOptional.get();
document.setEmail(event.getNewEmail());
@@ -125,9 +122,9 @@ public class BankAccountMongoProjection implements Projection {
@NewSpan
private void handle(@SpanTag("event") AddressUpdatedEvent event) {
log.info("(when) AddressUpdatedEvent: {}, aggregateID: {}", event, event.getAggregateId());
final var documentOptional = mongoRepository.findByAggregateId(event.getAggregateId());
Optional<BankAccountDocument> documentOptional = mongoRepository.findByAggregateId(event.getAggregateId());
if (documentOptional.isEmpty())
throw new BankAccountDocumentNotFoundException(event.getAggregateId());
throw new RuntimeException("Bank Account Document not found id: {}" + event.getAggregateId());
final var document = documentOptional.get();
document.setAddress(event.getNewAddress());
@@ -137,12 +134,13 @@ public class BankAccountMongoProjection implements Projection {
@NewSpan
private void handle(@SpanTag("event") BalanceDepositedEvent event) {
log.info("(when) BalanceDepositedEvent: {}, aggregateID: {}", event, event.getAggregateId());
final var documentOptional = mongoRepository.findByAggregateId(event.getAggregateId());
Optional<BankAccountDocument> documentOptional = mongoRepository.findByAggregateId(event.getAggregateId());
if (documentOptional.isEmpty())
throw new BankAccountDocumentNotFoundException(event.getAggregateId());
throw new RuntimeException("Bank Account Document not found id: {}" + event.getAggregateId());
final var document = documentOptional.get();
final var newBalance = document.getBalance().add(event.getAmount());
final var balance = document.getBalance();
final var newBalance = balance.add(event.getAmount());
document.setBalance(newBalance);
mongoRepository.save(document);
}

View File

@@ -6,15 +6,10 @@ import com.eventsourcing.bankAccount.dto.BankAccountResponseDTO;
import com.eventsourcing.bankAccount.repository.BankAccountMongoRepository;
import com.eventsourcing.es.EventStoreDB;
import com.eventsourcing.mappers.BankAccountMapper;
import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import io.github.resilience4j.retry.annotation.Retry;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.sleuth.annotation.NewSpan;
import org.springframework.cloud.sleuth.annotation.SpanTag;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.stereotype.Service;
import java.util.Optional;
@@ -27,12 +22,9 @@ public class BankAccountQueryHandler implements BankAccountQueryService {
private final EventStoreDB eventStoreDB;
private final BankAccountMongoRepository mongoRepository;
private static final String SERVICE_NAME = "microservice";
@Override
@NewSpan
@Retry(name = SERVICE_NAME)
@CircuitBreaker(name = SERVICE_NAME)
public BankAccountResponseDTO handle(@SpanTag("query") GetBankAccountByIDQuery query) {
Optional<BankAccountDocument> optionalDocument = mongoRepository.findByAggregateId(query.aggregateID());
if (optionalDocument.isPresent()) {
@@ -40,20 +32,11 @@ public class BankAccountQueryHandler implements BankAccountQueryService {
}
final var aggregate = eventStoreDB.load(query.aggregateID(), BankAccountAggregate.class);
final var savedDocument = mongoRepository.save(BankAccountMapper.bankAccountDocumentFromAggregate(aggregate));
BankAccountDocument savedDocument = mongoRepository.save(BankAccountMapper.bankAccountDocumentFromAggregate(aggregate));
log.info("(GetBankAccountByIDQuery) savedDocument: {}", savedDocument);
final var bankAccountResponseDTO = BankAccountMapper.bankAccountResponseDTOFromAggregate(aggregate);
log.info("(GetBankAccountByIDQuery) response: {}", bankAccountResponseDTO);
return bankAccountResponseDTO;
}
@Override
@NewSpan
@Retry(name = SERVICE_NAME)
@CircuitBreaker(name = SERVICE_NAME)
public Page<BankAccountResponseDTO> handle(@SpanTag("query") FindAllOrderByBalance query) {
return mongoRepository.findAll(PageRequest.of(query.page(), query.size(), Sort.by("balance")))
.map(BankAccountMapper::bankAccountResponseDTOFromDocument);
}
}

View File

@@ -1,9 +1,7 @@
package com.eventsourcing.bankAccount.queries;
import com.eventsourcing.bankAccount.dto.BankAccountResponseDTO;
import org.springframework.data.domain.Page;
public interface BankAccountQueryService {
BankAccountResponseDTO handle(GetBankAccountByIDQuery query);
Page<BankAccountResponseDTO> handle(FindAllOrderByBalance query);
}

View File

@@ -1,4 +0,0 @@
package com.eventsourcing.bankAccount.queries;
public record FindAllOrderByBalance(int page, int size) {
}

View File

@@ -28,6 +28,7 @@ public class KafkaProducerConfig {
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
producerProps.put(ProducerConfig.ACKS_CONFIG, kafkaConfigProperties.getAcks());
// producerProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, kafkaConfigProperties.getCompressionType());
producerProps.put(ProducerConfig.RETRIES_CONFIG, kafkaConfigProperties.getRetries());
producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, kafkaConfigProperties.getDeliveryTimeoutMs());
producerProps.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, kafkaConfigProperties.getMaxRequestSize());

View File

@@ -62,6 +62,10 @@ public abstract class AggregateRoot {
this.clearChanges();
}
public String string() {
return String.format("id: {%s}, type: {%s}, version: {%s}, changes: {%s}", id, type, version, changes.size());
}
private void validateEvent(final Event event) {
if (Objects.isNull(event) || !event.getAggregateId().equals(this.id))
throw new InvalidEventException(event.toString());

View File

@@ -6,6 +6,7 @@ import lombok.Data;
import lombok.NoArgsConstructor;
import java.time.LocalDateTime;
import java.time.ZonedDateTime;
import java.util.UUID;
@Data
@@ -13,15 +14,6 @@ import java.util.UUID;
@AllArgsConstructor
@Builder
public class Event {
private UUID id;
private String aggregateId;
private String eventType;
private String aggregateType;
private long version;
private byte[] data;
private byte[] metaData;
private LocalDateTime timeStamp;
public Event(String eventType, String aggregateType) {
this.id = UUID.randomUUID();
@@ -30,6 +22,23 @@ public class Event {
this.timeStamp = LocalDateTime.now();
}
private UUID id;
private String aggregateId;
private String eventType;
private String aggregateType;
private long version;
private byte[] data;
private byte[] metaData;
// @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd HH:mm")
private LocalDateTime timeStamp;
@Override
public String toString() {

View File

@@ -11,11 +11,8 @@ import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.stereotype.Repository;
import org.springframework.transaction.annotation.Transactional;
import java.io.Serializable;
import java.util.*;
import static com.eventsourcing.es.Constants.*;
@Repository
@RequiredArgsConstructor
@Slf4j
@@ -32,6 +29,57 @@ public class EventStore implements EventStoreDB {
private final NamedParameterJdbcTemplate jdbcTemplate;
private final EventBus eventBus;
@Override
@NewSpan
public void saveEvents(@SpanTag("events") List<Event> events) {
if (events.isEmpty()) return;
final List<Event> changes = new ArrayList<>(events);
changes.forEach(event -> {
int result = jdbcTemplate.update(SAVE_EVENTS_QUERY, Map.of(
"aggregate_id", event.getAggregateId(),
"aggregate_type", event.getAggregateType(),
"event_type", event.getEventType(),
"data", Objects.isNull(event.getData()) ? new byte[]{} : event.getData(),
"metadata", Objects.isNull(event.getMetaData()) ? new byte[]{} : event.getMetaData(),
"version", event.getVersion()));
log.info("(saveEvents) saved result: {}, event: {}", result, event);
});
}
@Override
@NewSpan
public List<Event> loadEvents(@SpanTag("aggregateId") String aggregateId, @SpanTag("version") long version) {
final List<Event> events = jdbcTemplate.query(LOAD_EVENTS_QUERY, Map.of("aggregate_id", aggregateId, "version", version),
(rs, rowNum) -> Event.builder()
.aggregateId(rs.getString("aggregate_id"))
.aggregateType(rs.getString("aggregate_type"))
.eventType(rs.getString("event_type"))
.data(rs.getBytes("data"))
.metaData(rs.getBytes("metadata"))
.version(rs.getLong("version"))
.timeStamp(rs.getTimestamp("timestamp").toLocalDateTime())
.build());
log.info("(loadEvents) events list: {}", events);
return events;
}
@NewSpan
private <T extends AggregateRoot> void saveSnapshot(@SpanTag("aggregate") T aggregate) {
aggregate.toSnapshot();
final var snapshot = EventSourcingUtils.snapshotFromAggregate(aggregate);
int update = jdbcTemplate.update(SAVE_SNAPSHOT_QUERY,
Map.of("aggregate_id", snapshot.getAggregateId(),
"aggregate_type", snapshot.getAggregateType(),
"data", Objects.isNull(snapshot.getData()) ? new byte[]{} : snapshot.getData(),
"metadata", Objects.isNull(snapshot.getMetaData()) ? new byte[]{} : snapshot.getMetaData(),
"version", snapshot.getVersion()));
log.info("(saveSnapshot) result: {}", update);
}
@Override
@Transactional
@@ -53,6 +101,50 @@ public class EventStore implements EventStoreDB {
log.info("(save) saved aggregate: {}", aggregate);
}
@NewSpan
private void handleConcurrency(@SpanTag("aggregateId") String aggregateId) {
try {
String aggregateID = jdbcTemplate.queryForObject(HANDLE_CONCURRENCY_QUERY, Map.of("aggregate_id", aggregateId), String.class);
log.info("(handleConcurrency) aggregateID for lock: {}", aggregateID);
} catch (EmptyResultDataAccessException e) {
log.info("(handleConcurrency) EmptyResultDataAccessException: {}", e.getMessage());
}
log.info("(handleConcurrency) aggregateID for lock: {}", aggregateId);
}
@NewSpan
private Optional<Snapshot> loadSnapshot(@SpanTag("aggregateId") String aggregateId) {
final Optional<Snapshot> snapshot = jdbcTemplate.query(LOAD_SNAPSHOT_QUERY, Map.of("aggregate_id", aggregateId), (rs, rowNum) -> Snapshot.builder()
.aggregateId(rs.getString("aggregate_id"))
.aggregateType(rs.getString("aggregate_type"))
.data(rs.getBytes("data"))
.metaData(rs.getBytes("metadata"))
.version(rs.getLong("version"))
.timeStamp(rs.getTimestamp("timestamp").toLocalDateTime())
.build()).stream().findFirst();
snapshot.ifPresent(result -> log.info("(loadSnapshot) snapshot: {}", result));
return snapshot;
}
@NewSpan
private <T extends AggregateRoot> T getAggregate(@SpanTag("aggregateId") final String aggregateId, @SpanTag("aggregateType") final Class<T> aggregateType) {
try {
return aggregateType.getConstructor(String.class).newInstance(aggregateId);
} catch (Exception ex) {
throw new RuntimeException(ex);
}
}
@NewSpan
private <T extends AggregateRoot> T getSnapshotFromClass(@SpanTag("snapshot") Optional<Snapshot> snapshot, @SpanTag("aggregateId") String aggregateId, @SpanTag("aggregateType") Class<T> aggregateType) {
if (snapshot.isEmpty()) {
final var defaultSnapshot = EventSourcingUtils.snapshotFromAggregate(getAggregate(aggregateId, aggregateType));
return EventSourcingUtils.aggregateFromSnapshot(defaultSnapshot, aggregateType);
}
return EventSourcingUtils.aggregateFromSnapshot(snapshot.get(), aggregateType);
}
@Override
@Transactional(readOnly = true)
@NewSpan
@@ -74,124 +166,16 @@ public class EventStore implements EventStoreDB {
return aggregate;
}
@Override
@NewSpan
public void saveEvents(@SpanTag("events") List<Event> events) {
if (events.isEmpty()) return;
final List<Event> changes = new ArrayList<>(events);
if (changes.size() > 1) {
this.eventsBatchInsert(changes);
return;
}
final Event event = changes.get(0);
int result = jdbcTemplate.update(SAVE_EVENTS_QUERY, mapFromEvent(event));
log.info("(saveEvents) saved result: {}, event: {}", result, event);
}
private Map<String, Serializable> mapFromEvent(Event event) {
return Map.of(
AGGREGATE_ID, event.getAggregateId(),
AGGREGATE_TYPE, event.getAggregateType(),
EVENT_TYPE, event.getEventType(),
DATA, Objects.isNull(event.getData()) ? new byte[]{} : event.getData(),
METADATA, Objects.isNull(event.getMetaData()) ? new byte[]{} : event.getMetaData(),
VERSION, event.getVersion());
}
@NewSpan
private void eventsBatchInsert(@SpanTag("events") List<Event> events) {
final var args = events.stream().map(this::mapFromEvent).toList();
final Map<String, ?>[] maps = args.toArray(new Map[0]);
int[] ints = jdbcTemplate.batchUpdate(SAVE_EVENTS_QUERY, maps);
log.info("(saveEvents) BATCH saved result: {}, event: {}", ints);
}
@Override
@NewSpan
public List<Event> loadEvents(@SpanTag("aggregateId") String aggregateId, @SpanTag("version") long version) {
return jdbcTemplate.query(LOAD_EVENTS_QUERY, Map.of(AGGREGATE_ID, aggregateId, VERSION, version),
(rs, rowNum) -> Event.builder()
.aggregateId(rs.getString(AGGREGATE_ID))
.aggregateType(rs.getString(AGGREGATE_TYPE))
.eventType(rs.getString(EVENT_TYPE))
.data(rs.getBytes(DATA))
.metaData(rs.getBytes(METADATA))
.version(rs.getLong(VERSION))
.timeStamp(rs.getTimestamp(TIMESTAMP).toLocalDateTime())
.build());
}
@NewSpan
private <T extends AggregateRoot> void saveSnapshot(@SpanTag("aggregate") T aggregate) {
aggregate.toSnapshot();
final var snapshot = EventSourcingUtils.snapshotFromAggregate(aggregate);
int updateResult = jdbcTemplate.update(SAVE_SNAPSHOT_QUERY,
Map.of(AGGREGATE_ID, snapshot.getAggregateId(),
AGGREGATE_TYPE, snapshot.getAggregateType(),
DATA, Objects.isNull(snapshot.getData()) ? new byte[]{} : snapshot.getData(),
METADATA, Objects.isNull(snapshot.getMetaData()) ? new byte[]{} : snapshot.getMetaData(),
VERSION, snapshot.getVersion()));
log.info("(saveSnapshot) updateResult: {}", updateResult);
}
@NewSpan
private void handleConcurrency(@SpanTag("aggregateId") String aggregateId) {
try {
String aggregateID = jdbcTemplate.queryForObject(HANDLE_CONCURRENCY_QUERY, Map.of(AGGREGATE_ID, aggregateId), String.class);
log.info("(handleConcurrency) aggregateID for lock: {}", aggregateID);
} catch (EmptyResultDataAccessException e) {
log.info("(handleConcurrency) EmptyResultDataAccessException: {}", e.getMessage());
}
log.info("(handleConcurrency) aggregateID for lock: {}", aggregateId);
}
@NewSpan
private Optional<Snapshot> loadSnapshot(@SpanTag("aggregateId") String aggregateId) {
return jdbcTemplate.query(LOAD_SNAPSHOT_QUERY, Map.of(AGGREGATE_ID, aggregateId), (rs, rowNum) -> Snapshot.builder()
.aggregateId(rs.getString(AGGREGATE_ID))
.aggregateType(rs.getString(AGGREGATE_TYPE))
.data(rs.getBytes(DATA))
.metaData(rs.getBytes(METADATA))
.version(rs.getLong(VERSION))
.timeStamp(rs.getTimestamp(TIMESTAMP).toLocalDateTime())
.build()).stream().findFirst();
}
@NewSpan
private <T extends AggregateRoot> T getAggregate(@SpanTag("aggregateId") final String aggregateId, @SpanTag("aggregateType") final Class<T> aggregateType) {
try {
return aggregateType.getConstructor(String.class).newInstance(aggregateId);
} catch (Exception ex) {
throw new RuntimeException(ex);
}
}
@NewSpan
private <T extends AggregateRoot> T getSnapshotFromClass(@SpanTag("snapshot") Optional<Snapshot> snapshot, @SpanTag("aggregateId") String aggregateId, @SpanTag("aggregateType") Class<T> aggregateType) {
if (snapshot.isEmpty()) {
final var defaultSnapshot = EventSourcingUtils.snapshotFromAggregate(getAggregate(aggregateId, aggregateType));
return EventSourcingUtils.aggregateFromSnapshot(defaultSnapshot, aggregateType);
}
return EventSourcingUtils.aggregateFromSnapshot(snapshot.get(), aggregateType);
}
@Override
@NewSpan
public Boolean exists(@SpanTag("aggregateId") String aggregateId) {
try {
final var id = jdbcTemplate.queryForObject(EXISTS_QUERY, Map.of(AGGREGATE_ID, aggregateId), String.class);
final var id = jdbcTemplate.queryForObject(EXISTS_QUERY, Map.of("aggregate_id", aggregateId), String.class);
log.info("aggregate exists id: {}", id);
return true;
} catch (Exception ex) {
if (!(ex instanceof EmptyResultDataAccessException)) {
throw new RuntimeException(ex);
throw new RuntimeException("exists", ex);
}
return false;
}

View File

@@ -17,9 +17,7 @@ import java.util.concurrent.TimeUnit;
@Slf4j
@RequiredArgsConstructor
public class KafkaEventBus implements EventBus {
private final KafkaTemplate<String, byte[]> kafkaTemplate;
private final static long sendTimeout = 3000;
@Value(value = "${order.kafka.topics.bank-account-event-store:bank-account-event-store}")
private String bankAccountTopicName;
@@ -31,11 +29,11 @@ public class KafkaEventBus implements EventBus {
final ProducerRecord<String, byte[]> record = new ProducerRecord<>(bankAccountTopicName, eventsBytes);
try {
kafkaTemplate.send(record).get(sendTimeout, TimeUnit.MILLISECONDS);
kafkaTemplate.send(record).get(3000, TimeUnit.MILLISECONDS);
log.info("publishing kafka record value >>>>> {}", new String(record.value()));
} catch (Exception ex) {
log.error("(KafkaEventBus) publish get timeout", ex);
log.error("(KafkaEventBus) publish get", ex);
throw new RuntimeException(ex);
}
}

View File

@@ -15,13 +15,21 @@ import java.util.UUID;
public class Snapshot {
private UUID id;
private String aggregateId;
private String aggregateType;
private byte[] data;
private byte[] metaData;
private long version;
private LocalDateTime timeStamp;
@Override
public String toString() {
return "Snapshot{" +

View File

@@ -1,6 +0,0 @@
package com.eventsourcing.exceptions;
import java.time.LocalDateTime;
public record ExceptionResponseDTO(int Status, String message, LocalDateTime timestamp) {
}

View File

@@ -0,0 +1,6 @@
package com.eventsourcing.exceptions;
import java.time.LocalDateTime;
public record NotFoundResponseDTO(int Status, String message, LocalDateTime timestamp) {
}

View File

@@ -1,11 +1,8 @@
package com.eventsourcing.filters;
import com.eventsourcing.bankAccount.exceptions.BankAccountDocumentNotFoundException;
import com.eventsourcing.bankAccount.exceptions.InvalidAddressException;
import com.eventsourcing.bankAccount.exceptions.InvalidEmailException;
import com.eventsourcing.es.exceptions.AggregateNotFoundException;
import com.eventsourcing.exceptions.InternalServerErrorResponse;
import com.eventsourcing.exceptions.ExceptionResponseDTO;
import com.eventsourcing.exceptions.NotFoundResponseDTO;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.annotation.Order;
@@ -28,6 +25,7 @@ import java.util.Map;
@Order(2)
public class GlobalControllerAdvice {
@ExceptionHandler(RuntimeException.class)
public ResponseEntity<InternalServerErrorResponse> handleRuntimeException(RuntimeException ex, WebRequest request) {
final var response = new InternalServerErrorResponse(HttpStatus.INTERNAL_SERVER_ERROR.value(), ex.getMessage(), LocalDateTime.now().toString());
@@ -46,18 +44,10 @@ public class GlobalControllerAdvice {
}
@ResponseStatus(HttpStatus.NOT_FOUND)
@ExceptionHandler({AggregateNotFoundException.class, BankAccountDocumentNotFoundException.class})
public ResponseEntity<ExceptionResponseDTO> handleAggregateNotFoundExceptions(AggregateNotFoundException ex) {
final var notFoundResponseDTO = new ExceptionResponseDTO(HttpStatus.NOT_FOUND.value(), ex.getMessage(), LocalDateTime.now());
log.error("handleAggregateNotFoundExceptions response ex:", ex);
@ExceptionHandler(AggregateNotFoundException.class)
public ResponseEntity<NotFoundResponseDTO> handleAggregateNotFoundException(AggregateNotFoundException ex) {
final var notFoundResponseDTO = new NotFoundResponseDTO(HttpStatus.NOT_FOUND.value(), ex.getMessage(), LocalDateTime.now());
log.error("AggregateNotFoundException response ex:", ex);
return ResponseEntity.status(HttpStatus.NOT_FOUND).body(notFoundResponseDTO);
}
@ResponseStatus(HttpStatus.BAD_REQUEST)
@ExceptionHandler({InvalidAddressException.class, InvalidEmailException.class})
public ResponseEntity<ExceptionResponseDTO> handleInvalidAggregateExceptions(AggregateNotFoundException ex) {
final var notFoundResponseDTO = new ExceptionResponseDTO(HttpStatus.BAD_REQUEST.value(), ex.getMessage(), LocalDateTime.now());
log.error("handleInvalidAggregateExceptions response ex:", ex);
return ResponseEntity.status(HttpStatus.BAD_REQUEST).body(notFoundResponseDTO);
}
}

View File

@@ -40,36 +40,7 @@ spring.data.mongodb.database=microservices
springdoc.swagger-ui.path=/swagger-ui.html
management.endpoints.web.exposure.include=health,prometheus,info
management.metrics.export.prometheus.enabled=true
spring.sleuth.propagation.type=w3c,b3
spring.sleuth.opentracing.enabled=true
spring.zipkin.base-url=http://localhost:9411
resilience4j.retry.instances.microservice.max-attempts=3
resilience4j.retry.instances.microservice.waitDuration=1s
resilience4j.retry.instances.microservice.enableExponentialBackoff=true
resilience4j.retry.instances.microservice.exponentialBackoffMultiplier=2
resilience4j.retry.instances.microservice.ignore-exceptions=com.eventsourcing.es.exceptions.AggregateNotFoundException
resilience4j.circuitbreaker.instances.microservice.registerHealthIndicator=true
resilience4j.circuitbreaker.instances.microservice.slidingWindowSize=5
resilience4j.circuitbreaker.instances.microservice.permittedNumberOfCallsInHalfOpenState=3
resilience4j.circuitbreaker.instances.microservice.slidingWindowType=TIME_BASED
resilience4j.circuitbreaker.instances.microservice.minimumNumberOfCalls=10
resilience4j.circuitbreaker.instances.microservice.waitDurationInOpenState=20s
resilience4j.circuitbreaker.instances.microservice.failureRateThreshold=30
resilience4j.circuitbreaker.instances.microservice.eventConsumerBufferSize=10
resilience4j.thread-pool-bulkhead.instances.microservice.maxThreadPoolSize=1
resilience4j.thread-pool-bulkhead.instances.microservice.coreThreadPoolSize=1
resilience4j.thread-pool-bulkhead.instances.microservice.queueCapacity=1
resilience4j.timelimiter.instances.microservice.timeoutDuration=3s
resilience4j.timelimiter.instances.microservice.cancelRunningFuture=true
spring.jdbc.template.query-timeout=3
spring.zipkin.base-url=http://localhost:9411