eventuate-tram / eventuate-tram-examples-customers-and-orders Goto Github PK
View Code? Open in Web Editor NEWAn example of Choreography-based sagas in Spring Boot/JPA microservices
License: Other
An example of Choreography-based sagas in Spring Boot/JPA microservices
License: Other
JSonMapper.fromJson(restTemplate.postForObject(baseUrlCustomers("customers/make-snapshot"), null, String.class), TopicPartitionOffset[].class
First, this should be simplified to
restTemplate.postForObject(baseUrlCustomers("customers/make-snapshot"), null, TopicPartitionOffset[].class)
also it is better to wrap arrays to object
@Autowired
FeignClient client;
public Message consumption_must_success_eventually(CommandMessage<RemoteActionCommand> cm) {
....
// it may throw an exception (like a momentary network glitch)
client.invokeRpc();
withSuccess();
}
Thanks!
e.g. Doesn't the CDC have sensible defaults for polling?
SPRING_DATASOURCE_TEST_ON_BORROW: "true"
SPRING_DATASOURCE_VALIDATION_QUERY: SELECT 1
EVENTUATELOCAL_KAFKA_BOOTSTRAP_SERVERS: kafka:29092
EVENTUATELOCAL_ZOOKEEPER_CONNECTION_STRING: zookeeper:2181
EVENTUATELOCAL_CDC_POLLING_INTERVAL_IN_MILLISECONDS: 500
EVENTUATELOCAL_CDC_MAX_EVENTS_PER_POLLING: 1000
EVENTUATELOCAL_CDC_MAX_ATTEMPTS_FOR_POLLING: 100
EVENTUATELOCAL_CDC_POLLING_RETRY_INTERVAL_IN_MILLISECONDS: 500
Tag with development-latest?
Let's say orderservice
crashed and lost all the data (not in this example, since order and customer share same db here), how can we replay all the events and restore its latest state?
In io.eventuate.local.db.log.common.DbLogBasedCdcProcessor
,
public void start(Consumer<EVENT> eventConsumer) {
Optional<BinlogFileOffset> startingBinlogFileOffset = offsetStore.getLastBinlogFileOffset();
process(eventConsumer, startingBinlogFileOffset);
}
The offsetStore
always seeks to the latest offset even though its consumer starts at the earliest
.
I guess we can either make it configurable or provide a different implementation of the CdcProcessor
.
But the question is, while we replay all the events, the orderservice
will emit new events as well, and those events are already published and stored in the event store. So, what do we do? publish them to a separated Topic? What's the recommended operation? Or, is it really practical?
I am using IntelliJ. I tried both SDK corretto-1.8 and openjdk-17
I receive the following build failure:
FAILURE: Build failed with an exception.
* What went wrong:
Execution failed for task ':buildSrc:compileGroovy'.
> BUG! exception in phase 'semantic analysis' in source unit 'C:\GitHub\eventuate-tram\eventuate-tram-examples-customers-and-orders\buildSrc\src\main\groovy\ServicePlugin.groovy' Unsupported class file major version 61
Needs multi-architecture base images.
For example: eventuate-foundation/eventuate-common#106
In this project, I think that the CDC service and Kafka provided will not guarantee the order for commands with overlapping sequences due to the two Kafka partitions. Is there a way to ensure order?
Building/Rebuilding CQRS views
Problem
Let’s imagine that the Eventuate Tram Customers and Orders project was running in production. You then wanted to add the Order History View - #1
You need to populate the order history views - Customer and Order MongoDB tables - with the existing data.
Assumptions
Let’s focus on Kafka for now
Solution
Note:
The goal is to build a framework that makes it easier to implement this behavior for any service/application
In the meantime, let’s just focus on the Eventuate Tram customer and orders application
Exporting data from a service
A service must expose an API for triggering the export behavior.
An export consists of an AggregateSnapshot event, which is an event containing a snapshot of the aggregate’s state.
A service (Order Service, Customer Service) exports its data as follows:
Lock table
This ensures that any domain events representing an update to an aggregate will follows the aggregate’s Order/CustomerSnapshotEvent
Publish StartOfSnapshots events to Order/Customer channel (topic) to indicate starting offset
Since the table is locked, the offset is the starting point for the consumer to read the Snapshot events.
Publish one StartOfSnapshots event to each partition in the topic using round robin with a null message key
Iterate through Orders/Customers and publish an (Customer/Order)Snapshot event for each one to the Order/Customer
The event containers the complete state of the Order/Customer
Perhaps one day use Spring batch
Unsure whether this is needed
Publish End Of Snapshots events to channel to indicate done
Unlock table
Importing data into a CQRS view service
The OrderHistoryService subscribes to (Customer/Order)Snapshot and updates the view appropriately.
One all of the services (Customer and Order) have published their StartOfSnapshots events and the Kafka consumer offsets are known:
Change the kafka consumer offsets for the OrderHistoryService’s subscription to the specified point offsets
Start the OrderHistoryService
It's too error prone otherwise.
And
eventuate-tram-examples-customers-and-orders/pom.xml
Lines 16 to 18 in 4d5206b
step to reproduce:
expected:
[INFO] order-history-backend .............................. FAILURE [ 2.519 s]
The same when doing it via idea.
Caused by: org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.springframework.test.context.cache.DefaultCacheAwareContextLoaderDelegate]: Unresolvable class definition; nested exception is java.lang.NoClassDefFoundError: org/springframework/core/KotlinDetector
e.g. create a job that runs mcr.microsoft.com/mssql-tools
and has it execute a script that downloads and executes the SQL scripts.
Terraform executes the job: https://www.terraform.io/docs/providers/kubernetes/r/job.html
Goal: ensure that the example applications use resources effectively.
This needs to be done for all examples, especially eventuate-*-customers-and-orders
docker image differences along with
targetCompatibility = '8'
sourceCompatibility = '8'
can just be parameterized ./gradlew -P x=y
There is no need for a branch. CircleCI can have a job that installs java 14 and builds with these parameters.
Reason: build-and-test-snapshots.sh failed
*Reason: Failed io.eventuate.examples.tram.ordersandcustomers.snapshottests.SnapshotTest testCustomers
**Reason: I/O error on GET request for "http://localhost:8084/customers": Connection refused (Connection refused); nested exception is java.net.ConnectException: Connection refused (Connection refused)
***Reason: docker-compose failed to start eventuate-tram-examples-customers-and-orders_order-history-text-search-service_2
****Reason: ERROR: for eventuate-tram-examples-customers-and-orders_order-history-text-search-service_2 Cannot start service order-history-text-search-service: driver failed programming external connectivity on endpoint eventuate-tram-examples-customers-and-orders_order-history-text-search-service_2 (0a3fca82cb3f008c791cb248399a5dfa7b15c30f38e70bc8ccbba1b0621cf112): Bind for 0.0.0.0:8084 failed: port is already allocated
*****Reason: there is already eventuate-tram-examples-customers-and-orders_order-history-text-search-service_1
******Reason: order-history-text-search-service starts by ordersTest and customersTest. ordersTest executed first. It tried to start order-history-text-search-service, which depends on elastic search, and while elastic search is pulled, order-history-text-search-service started timeout is elapsed. So, customersTest started creating order-history-text-search-service concurrently
Solution: Increase timeouts start elastic search before order-history-text-search-service
Java Version:
PS C:\GitHub\eventuate-tram\eventuate-tram-examples-customers-and-orders> java -version openjdk version "1.8.0_322" OpenJDK Runtime Environment Corretto-8.322.06.1 (build 1.8.0_322-b06) OpenJDK 64-Bit Server VM Corretto-8.322.06.1 (build 25.322-b06, mixed mode) PS C:\GitHub\eventuate-tram\eventuate-tram-examples-customers-and-orders>
Gradle Build:
`
PS C:\GitHub\eventuate-tram\eventuate-tram-examples-customers-and-orders> ./gradlew
BUILD SUCCESSFUL in 881ms
1 actionable task: 1 executed
`
./gradlew buildAndStartServicesMySql
`
o.s.boot.SpringApplication : Application run failed
org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'eventuateKafkaProducer' defined in class pa
th resource [io/eventuate/tram/cdc/connector/configuration/KafkaMessageTableChangesToDestinationsConfiguration.class]: Bean instantia
tion via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [io.e
ventuate.messaging.kafka.producer.EventuateKafkaProducer]: Factory method 'eventuateKafkaProducer' threw exception; nested exception is org.apache.kafka.common.KafkaException: Failed to construct kafka producer
at
`
See the attached log file:
Need to retry DuplicateKeyException thrown by upserts.
This occurred when deploying on Fargate with DocumentDB:
org.springframework.dao.DuplicateKeyException: E11000 duplicate key error collection: customerView index: _id_; nested exception is com.mongodb.MongoWriteException: E11000 duplicate key error collection: customerView index: _id_
at org.springframework.data.mongodb.core.MongoExceptionTranslator.translateExceptionIfPossible(MongoExceptionTranslator.java:101)
at org.springframework.data.mongodb.core.MongoTemplate.potentiallyConvertRuntimeException(MongoTemplate.java:2781)
at org.springframework.data.mongodb.core.MongoTemplate.execute(MongoTemplate.java:547)
at org.springframework.data.mongodb.core.MongoTemplate.doUpdate(MongoTemplate.java:1578)
at org.springframework.data.mongodb.core.MongoTemplate.doUpdate(MongoTemplate.java:1568)
at org.springframework.data.mongodb.core.MongoTemplate.upsert(MongoTemplate.java:1514)
at io.eventuate.examples.tram.ordersandcustomers.orderhistory.backend.CustomerViewRepositoryImpl.addOrder(CustomerViewRepositoryImpl.java:31)
Apparently, this is a feature:
Problem:
Eventuate cdc service caches primary keys:
After db id migration primary key column is changed.
New message records after migration have empty old primary key.
When any new message processed, all new records marked as published,
because have empty old primary key (and cdc uses it when marks event published), so some messages were ignored by cdc.
Solution:
Restart cdc service after migration.
Is this a typo (or just a confusing ordering, appearing just below the creation of the "entities" table, but adding an index to a different table)?
CREATE INDEX entities_idx ON events(entity_type, entity_id);
Unclear why this started to happen with 884c4ba:
Eventuate CDC connects without issue.
2023-10-15 17:33:29.089 INFO [customer-service,,] 1 --- [ main] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Starting...
2023-10-15 17:33:31.216 ERROR [customer-service,,] 1 --- [ main] com.zaxxer.hikari.pool.HikariPool : HikariPool-1 - Exception during pool initialization.
com.microsoft.sqlserver.jdbc.SQLServerException: The driver could not establish a secure connection to SQL Server by using Secure Sockets Layer (SSL) encryption. Error: "PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target". ClientConnectionId:f9546012-b5ed-4e53-801a-ba7b3449f429
I am getting the following error when running:
./gradlew mysqlbinlogComposeBuild mysqlbinlogComposeUp
Task :mysqlbinlogComposeBuild
Invalid interpolation format for "build" option in service "customerservice": "${EVENTUATE_JAVA_BASE_IMAGE_VERSION?}"
How must EVENTUATE_JAVA_BASE_IMAGE_VERSION be set?
Environment:
Linux Mint 20
Java: openjdk version "1.8.0_275"
Error in logs:
org.springframework.jdbc.BadSqlGrammarException: PreparedStatementCallback; bad SQL grammar [select top (?) * FROM eventuate.message WHERE published = 0 ORDER BY id ASC]; nested exception is com.microsoft.sqlserver.jdbc.SQLServerException: Invalid object name 'eventuate.message'.
at org.springframework.jdbc.support.SQLErrorCodeSQLExceptionTranslator.doTranslate(SQLErrorCodeSQLExceptionTranslator.java:234) ~[spring-jdbc-5.1.3.RELEASE.jar!/:5.1.3.RELEASE]
at org.springframework.jdbc.support.AbstractFallbackSQLExceptionTranslator.translate(AbstractFallbackSQLExceptionTranslator.java:72) ~[spring-jdbc-5.1.3.RELEASE.jar!/:5.1.3.RELEASE]
at org.springframework.jdbc.core.JdbcTemplate.translateException(JdbcTemplate.java:1444) ~[spring-jdbc-5.1.3.RELEASE.jar!/:5.1.3.RELEASE]
at org.springframework.jdbc.core.JdbcTemplate.execute(JdbcTemplate.java:632) ~[spring-jdbc-5.1.3.RELEASE.jar!/:5.1.3.RELEASE]
at org.springframework.jdbc.core.JdbcTemplate.query(JdbcTemplate.java:669) ~[spring-jdbc-5.1.3.RELEASE.jar!/:5.1.3.RELEASE]
at org.springframework.jdbc.core.JdbcTemplate.query(JdbcTemplate.java:694) ~[spring-jdbc-5.1.3.RELEASE.jar!/:5.1.3.RELEASE]
at org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate.queryForRowSet(NamedParameterJdbcTemplate.java:307) ~[spring-jdbc-5.1.3.RELEASE.jar!/:5.1.3.RELEASE]
at org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate.queryForRowSet(NamedParameterJdbcTemplate.java:315) ~[spring-jdbc-5.1.3.RELEASE.jar!/:5.1.3.RELEASE]
at io.eventuate.local.polling.PollingDao.lambda$processEvents$1(PollingDao.java:124) [eventuate-local-java-cdc-connector-polling-0.10.0.DB_ID_GEN.BUILD-SNAPSHOT.jar!/:na]
at io.eventuate.local.common.DaoUtils.handleConnectionLost(DaoUtils.java:22) ~[eventuate-local-java-cdc-connector-common-0.10.0.DB_ID_GEN.BUILD-SNAPSHOT.jar!/:na]
at io.eventuate.local.polling.PollingDao.processEvents(PollingDao.java:122) [eventuate-local-java-cdc-connector-polling-0.10.0.DB_ID_GEN.BUILD-SNAPSHOT.jar!/:na]
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193) ~[na:1.8.0_252]
at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948) ~[na:1.8.0_252]
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482) ~[na:1.8.0_252]
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472) ~[na:1.8.0_252]
at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708) ~[na:1.8.0_252]
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[na:1.8.0_252]
at java.util.stream.ReferencePipeline.reduce(ReferencePipeline.java:541) ~[na:1.8.0_252]
at io.eventuate.local.polling.PollingDao.start(PollingDao.java:97) [eventuate-local-java-cdc-connector-polling-0.10.0.DB_ID_GEN.BUILD-SNAPSHOT.jar!/:na]
at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_252]
Caused by: com.microsoft.sqlserver.jdbc.SQLServerException: Invalid object name 'eventuate.message'.
at com.microsoft.sqlserver.jdbc.SQLServerException.makeFromDatabaseError(SQLServerException.java:256) ~[mssql-jdbc-7.2.1.jre8.jar!/:na]
at com.microsoft.sqlserver.jdbc.SQLServerStatement.getNextResult(SQLServerStatement.java:1621) ~[mssql-jdbc-7.2.1.jre8.jar!/:na]
at com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement.doExecutePreparedStatement(SQLServerPreparedStatement.java:592) ~[mssql-jdbc-7.2.1.jre8.jar!/:na]
at com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement$PrepStmtExecCmd.doExecute(SQLServerPreparedStatement.java:522) ~[mssql-jdbc-7.2.1.jre8.jar!/:na]
at com.microsoft.sqlserver.jdbc.TDSCommand.execute(IOBuffer.java:7194) ~[mssql-jdbc-7.2.1.jre8.jar!/:na]
at com.microsoft.sqlserver.jdbc.SQLServerConnection.executeCommand(SQLServerConnection.java:2930) ~[mssql-jdbc-7.2.1.jre8.jar!/:na]
at com.microsoft.sqlserver.jdbc.SQLServerStatement.executeCommand(SQLServerStatement.java:248) ~[mssql-jdbc-7.2.1.jre8.jar!/:na]
at com.microsoft.sqlserver.jdbc.SQLServerStatement.executeStatement(SQLServerStatement.java:223) ~[mssql-jdbc-7.2.1.jre8.jar!/:na]
at com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement.executeQuery(SQLServerPreparedStatement.java:444) ~[mssql-jdbc-7.2.1.jre8.jar!/:na]
at com.zaxxer.hikari.pool.ProxyPreparedStatement.executeQuery(ProxyPreparedStatement.java:52) ~[HikariCP-3.2.0.jar!/:na]
at com.zaxxer.hikari.pool.HikariProxyPreparedStatement.executeQuery(HikariProxyPreparedStatement.java) ~[HikariCP-3.2.0.jar!/:na]
at org.springframework.jdbc.core.JdbcTemplate$1.doInPreparedStatement(JdbcTemplate.java:678) ~[spring-jdbc-5.1.3.RELEASE.jar!/:5.1.3.RELEASE]
at org.springframework.jdbc.core.JdbcTemplate.execute(JdbcTemplate.java:617) ~[spring-jdbc-5.1.3.RELEASE.jar!/:5.1.3.RELEASE]
... 16 common frames omitted
Prably reason is eventuate-foundation/eventuate-common#71
Now fails with the input device is not a TTY
here
docker-compose -f ${migration_tool} run --no-deps mssql-migration
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google ❤️ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.