BAEL-4467: Introduction to Debezium

2021-04-04 18:08:07 +02:00
parent 96e97ae79b
commit 082f3133ba
14 changed files with 488 additions and 3 deletions
@@ -0,0 +1,13 @@
+package com.baeldung.libraries.debezium;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class DebeziumCDCApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(DebeziumCDCApplication.class, args);
+    }
+
+}
@@ -0,0 +1,58 @@
+package com.baeldung.libraries.debezium.config;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.io.File;
+import java.io.IOException;
+
+@Configuration
+public class DebeziumConnectorConfig {
+
+    /**
+     * Database details.
+     */
+    @Value("${customer.datasource.host}")
+    private String customerDbHost;
+
+    @Value("${customer.datasource.database}")
+    private String customerDbName;
+
+    @Value("${customer.datasource.port}")
+    private String customerDbPort;
+
+    @Value("${customer.datasource.username}")
+    private String customerDbUsername;
+
+    @Value("${customer.datasource.password}")
+    private String customerDbPassword;
+
+    /**
+     * Customer Database Connector Configuration
+     */
+    @Bean
+    public io.debezium.config.Configuration customerConnector() throws IOException {
+        File offsetStorageTempFile = File.createTempFile("offsets_", ".dat");
+        File dbHistoryTempFile = File.createTempFile("dbhistory_", ".dat");
+        return io.debezium.config.Configuration.create()
+            .with("name", "customer-mysql-connector")
+            .with("connector.class", "io.debezium.connector.mysql.MySqlConnector")
+            .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
+            .with("offset.storage.file.filename", offsetStorageTempFile.getAbsolutePath())
+            .with("offset.flush.interval.ms", "60000")
+            .with("database.hostname", customerDbHost)
+            .with("database.port", customerDbPort)
+            .with("database.user", customerDbUsername)
+            .with("database.password", customerDbPassword)
+            .with("database.dbname", customerDbName)
+            .with("database.include.list", customerDbName)
+            .with("include.schema.changes", "false")
+            .with("database.allowPublicKeyRetrieval", "true")
+            .with("database.server.id", "10181")
+            .with("database.server.name", "customer-mysql-db-server")
+            .with("database.history", "io.debezium.relational.history.FileDatabaseHistory")
+            .with("database.history.file.filename", dbHistoryTempFile.getAbsolutePath())
+            .build();
+    }
+}
@@ -0,0 +1,18 @@
+package com.baeldung.libraries.debezium.entity;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import javax.persistence.Entity;
+import javax.persistence.Id;
+
+@Entity
+@Getter
+@Setter
+public class Customer {
+    @Id
+    private Long id;
+
+    private String fullname;
+    private String email;
+}
@@ -0,0 +1,83 @@
+package com.baeldung.libraries.debezium.listener;
+
+import com.baeldung.libraries.debezium.service.CustomerService;
+import io.debezium.config.Configuration;
+import io.debezium.embedded.Connect;
+import io.debezium.engine.DebeziumEngine;
+import io.debezium.engine.RecordChangeEvent;
+import io.debezium.engine.format.ChangeEventFormat;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.kafka.connect.data.Field;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+
+import static io.debezium.data.Envelope.FieldName.*;
+import static io.debezium.data.Envelope.Operation;
+import static java.util.stream.Collectors.toMap;
+
+@Slf4j
+@Component
+public class DebeziumListener {
+
+    private final Executor executor = Executors.newSingleThreadExecutor();
+    private final CustomerService customerService;
+    private final DebeziumEngine<RecordChangeEvent<SourceRecord>> debeziumEngine;
+
+    public DebeziumListener(Configuration customerConnectorConfiguration, CustomerService customerService) {
+
+        this.debeziumEngine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
+            .using(customerConnectorConfiguration.asProperties())
+            .notifying(this::handleChangeEvent)
+            .build();
+
+        this.customerService = customerService;
+    }
+
+    private void handleChangeEvent(RecordChangeEvent<SourceRecord> sourceRecordRecordChangeEvent) {
+        SourceRecord sourceRecord = sourceRecordRecordChangeEvent.record();
+
+        log.info("Key = '" + sourceRecord.key() + "' value = '" + sourceRecord.value() + "'");
+
+        Struct sourceRecordChangeValue= (Struct) sourceRecord.value();
+
+        if (sourceRecordChangeValue != null) {
+            Operation operation = Operation.forCode((String) sourceRecordChangeValue.get(OPERATION));
+
+            if(operation != Operation.READ) {
+                String record = operation == Operation.DELETE ? BEFORE : AFTER; // Handling Update & Insert operations.
+
+                Struct struct = (Struct) sourceRecordChangeValue.get(record);
+                Map<String, Object> payload = struct.schema().fields().stream()
+                    .map(Field::name)
+                    .filter(fieldName -> struct.get(fieldName) != null)
+                    .map(fieldName -> Pair.of(fieldName, struct.get(fieldName)))
+                    .collect(toMap(Pair::getKey, Pair::getValue));
+
+                this.customerService.replicateData(payload, operation);
+                log.info("Updated Data: {} with Operation: {}", payload, operation.name());
+            }
+        }
+    }
+
+    @PostConstruct
+    private void start() {
+        this.executor.execute(debeziumEngine);
+    }
+
+    @PreDestroy
+    private void stop() throws IOException {
+        if (this.debeziumEngine != null) {
+            this.debeziumEngine.close();
+        }
+    }
+
+}
@@ -0,0 +1,9 @@
+package com.baeldung.libraries.debezium.repository;
+
+import com.baeldung.libraries.debezium.entity.Customer;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface CustomerRepository extends JpaRepository<Customer, Long> {
+}
@@ -0,0 +1,30 @@
+package com.baeldung.libraries.debezium.service;
+
+import com.baeldung.libraries.debezium.entity.Customer;
+import com.baeldung.libraries.debezium.repository.CustomerRepository;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.debezium.data.Envelope.Operation;
+import org.springframework.stereotype.Service;
+
+import java.util.Map;
+
+@Service
+public class CustomerService {
+
+    private final CustomerRepository customerRepository;
+
+    public CustomerService(CustomerRepository customerRepository) {
+        this.customerRepository = customerRepository;
+    }
+
+    public void replicateData(Map<String, Object> customerData, Operation operation) {
+        final ObjectMapper mapper = new ObjectMapper();
+        final Customer customer = mapper.convertValue(customerData, Customer.class);
+
+        if (Operation.DELETE.name().equals(operation.name())) {
+            customerRepository.deleteById(customer.getId());
+        } else {
+            customerRepository.save(customer);
+        }
+    }
+}