From 0266f5be29af6fee0569c09fb559c8cfdfd5a3a3 Mon Sep 17 00:00:00 2001
From: Corneil du Plessis <pcorneil@vmware.com>
Date: Thu, 4 May 2023 16:19:27 +0200
Subject: [PATCH] Remove class from `DebeziumFormat`

---
 .../DebeziumEngineAutoConfiguration.java         | 15 +++++++++++++--
 .../fn/supplier/debezium/DebeziumProperties.java | 16 +++++-----------
 2 files changed, 18 insertions(+), 13 deletions(-)

diff --git a/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineAutoConfiguration.java b/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineAutoConfiguration.java
index 24453708..3b1dfa09 100644
--- a/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineAutoConfiguration.java
+++ b/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineAutoConfiguration.java
@@ -18,12 +18,15 @@ package org.springframework.cloud.fn.supplier.debezium;
 
 import java.time.Clock;
 import java.time.Duration;
+import java.util.Map;
+import java.util.Objects;
 import java.util.function.Consumer;
 
 import io.debezium.engine.ChangeEvent;
 import io.debezium.engine.DebeziumEngine;
 import io.debezium.engine.DebeziumEngine.CompletionCallback;
 import io.debezium.engine.DebeziumEngine.ConnectorCallback;
+import io.debezium.engine.format.SerializationFormat;
 import io.debezium.engine.spi.OffsetCommitPolicy;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -33,6 +36,8 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 
+import static org.springframework.cloud.fn.supplier.debezium.DebeziumProperties.DebeziumFormat;
+
 /**
  * DebeziumEngine auto-configuration.
  *
@@ -46,6 +51,7 @@ import org.springframework.context.annotation.Bean;
  * restart.
  *
  * @author Christian Tzolov
+ * @author Corneil du Plessis
  */
 @AutoConfiguration
 @EnableConfigurationProperties(DebeziumProperties.class)
@@ -105,14 +111,19 @@ public class DebeziumEngineAutoConfiguration {
 	public ConnectorCallback connectorCallback() {
 		return DEFAULT_CONNECTOR_CALLBACK;
 	}
-
+	private static final Map<DebeziumFormat, Class<? extends SerializationFormat<byte[]>>> serialFormats = Map.of(
+		DebeziumFormat.JSON, io.debezium.engine.format.JsonByteArray.class,
+		DebeziumFormat.AVRO, io.debezium.engine.format.Avro.class,
+		DebeziumFormat.PROTOBUF, io.debezium.engine.format.Protobuf.class
+	);
 	@Bean
 	public DebeziumEngine<?> debeziumEngine(Consumer<ChangeEvent<byte[], byte[]>> changeEventConsumer,
 			OffsetCommitPolicy offsetCommitPolicy, CompletionCallback completionCallback,
 			ConnectorCallback connectorCallback, DebeziumProperties properties, Clock debeziumClock) {
 
+		Class<? extends SerializationFormat<byte[]>> format = Objects.requireNonNull(serialFormats.get(properties.getFormat()), "Cannot find format for " + properties.getProperties());
 		DebeziumEngine<ChangeEvent<byte[], byte[]>> debeziumEngine = DebeziumEngine
-				.create(properties.getFormat().serializationFormat())
+				.create(format)
 				.using(properties.getDebeziumNativeConfiguration())
 				.using(debeziumClock)
 				.using(completionCallback)
diff --git a/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumProperties.java b/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumProperties.java
index 640ee9ca..9e544b06 100644
--- a/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumProperties.java
+++ b/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumProperties.java
@@ -20,8 +20,6 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
 
-import io.debezium.engine.format.SerializationFormat;
-
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
 /**
@@ -34,27 +32,23 @@ public class DebeziumProperties {
 		/**
 		 * JSON change event format.
 		 */
-		JSON(io.debezium.engine.format.JsonByteArray.class, "application/json"),
+		JSON("application/json"),
 		/**
 		 * AVRO change event format.
 		 */
-		AVRO(io.debezium.engine.format.Avro.class, "application/avro"),
+		AVRO("application/avro"),
 		/**
 		 * ProtoBuf change event format.
 		 */
-		PROTOBUF(io.debezium.engine.format.Protobuf.class, "application/x-protobuf"),;
+		PROTOBUF("application/x-protobuf"),;
+
 
-		private final Class<? extends SerializationFormat<byte[]>> serializationFormat;
 		private final String contentType;
 
-		DebeziumFormat(Class<? extends SerializationFormat<byte[]>> serializationFormat, String contentType) {
-			this.serializationFormat = serializationFormat;
+		DebeziumFormat(String contentType) {
 			this.contentType = contentType;
 		}
 
-		public Class<? extends SerializationFormat<byte[]>> serializationFormat() {
-			return serializationFormat;
-		}
 
 		public final String contentType() {
 			return contentType;