use flyway

cqrs-source/pom.xml

@@ -73,6 +73,12 @@
 			<scope>provided</scope>
 		</dependency>
 
+		<dependency>
+			<groupId>org.flywaydb</groupId>
+			<artifactId>flyway-core</artifactId>
+			<version>5.0.7</version>
+		</dependency>
+
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-test</artifactId>

cqrs-source/src/main/resources/application.properties

@@ -9,4 +9,12 @@ spring.jpa.properties.hibernate.dialect = org.hibernate.dialect.PostgreSQL9Diale
 
 spring.cloud.stream.kafka.binder.brokers=kafka:9092
 spring.cloud.stream.bindings.output.destination=events
-spring.cloud.stream.bindings.output.contentType=application/json
+spring.cloud.stream.bindings.output.contentType=application/json
+
+spring.flyway.enabled=true
+spring.flyway.check-location=true
+spring.flyway.locations=classpath:db/migration
+spring.flyway.schemas=cqrs
+spring.flyway.url=jdbc:postgresql://db:5432/mydb
+spring.flyway.user=myuser
+spring.flyway.password=mypwd

cqrs-source/src/main/resources/db/migration/V1.0__createTables.sql

@@ -0,0 +1,31 @@
+START TRANSACTION;
+
+CREATE SCHEMA IF NOT EXISTS cqrs;
+
+-- bank_account table is used by the source service
+CREATE TABLE cqrs.bank_account
+(
+  id SERIAL NOT NULL,
+  account_number CHARACTER VARYING(255) NOT NULL,
+  balance BIGINT,
+  CONSTRAINT bank_account_pkey PRIMARY KEY (id)
+)
+WITH (
+OIDS = FALSE
+);
+
+-- transaction table is used by the sink service
+CREATE TABLE cqrs.transaction
+(
+  id SERIAL NOT NULL,
+  account_number CHARACTER VARYING(255) NOT NULL,
+  type CHARACTER VARYING(20),
+  amount BIGINT,
+  creation_date TIMESTAMP WITHOUT TIME ZONE NOT NULL,
+  CONSTRAINT transaction_pkey PRIMARY KEY (id)
+)
+WITH (
+OIDS = FALSE
+);
+
+COMMIT;

cqrs-source/src/main/resources/db/migration/V1.1__dataInit.sql

@@ -0,0 +1,9 @@
+START TRANSACTION;
+
+INSERT INTO cqrs.bank_account (account_number, balance) VALUES
+  (101, 5000);
+
+INSERT INTO cqrs.bank_account (account_number, balance) VALUES
+  (102, 3000);
+
+COMMIT;