GH-8: Debezium auto-config: always use Json for headers format (#34)

Fixes: #8

It really does not make sense to keep headers in the `byte[]`, when typically all of them are strings.

* Change `DebeziumEngineBuilderAutoConfiguration` to always use `Json.class` for headers
* Remove `DebeziumProperties.headerFormat` respectively.

common/spring-debezium-autoconfigure/src/main/java/org/springframework/cloud/fn/common/debezium/DebeziumEngineBuilderAutoConfiguration.java

@@ -23,6 +23,7 @@
 import io.debezium.engine.DebeziumEngine;
 import io.debezium.engine.DebeziumEngine.CompletionCallback;
 import io.debezium.engine.DebeziumEngine.ConnectorCallback;
+import io.debezium.engine.format.Json;
 import io.debezium.engine.format.KeyValueHeaderChangeEventFormat;
 import io.debezium.engine.format.SerializationFormat;
 import io.debezium.engine.spi.OffsetCommitPolicy;
@@ -140,11 +141,7 @@ public DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder
 				serializationFormatClass(properties.getPayloadFormat()),
 				"Cannot find payload format for " + properties.getProperties());
 
-		Class<? extends SerializationFormat<byte[]>> headerFormat = Objects.requireNonNull(
-				serializationFormatClass(properties.getHeaderFormat()),
-				"Cannot find header format for " + properties.getProperties());
-
-		return DebeziumEngine.create(KeyValueHeaderChangeEventFormat.of(payloadFormat, payloadFormat, headerFormat))
+		return DebeziumEngine.create(KeyValueHeaderChangeEventFormat.of(payloadFormat, payloadFormat, Json.class))
 			.using(properties.getDebeziumNativeConfiguration())
 			.using(debeziumClock)
 			.using(completionCallback)

common/spring-debezium-autoconfigure/src/main/java/org/springframework/cloud/fn/common/debezium/DebeziumProperties.java

@@ -69,11 +69,6 @@ public final String contentType() {
 	 */
 	private DebeziumFormat payloadFormat = DebeziumFormat.JSON;
 
-	/**
-	 * {@code io.debezium.engine.ChangeEvent} header format. Defaults to 'JSON'.
-	 */
-	private DebeziumFormat headerFormat = DebeziumFormat.JSON;
-
 	/**
 	 * The policy that defines when the offsets should be committed to offset storage.
 	 */
@@ -91,14 +86,6 @@ public void setPayloadFormat(DebeziumFormat format) {
 		this.payloadFormat = format;
 	}
 
-	public DebeziumFormat getHeaderFormat() {
-		return this.headerFormat;
-	}
-
-	public void setHeaderFormat(DebeziumFormat headerFormat) {
-		this.headerFormat = headerFormat;
-	}
-
 	public enum DebeziumOffsetCommitPolicy {
 
 		/**

common/spring-debezium-autoconfigure/src/test/java/org/springframework/cloud/fn/common/debezium/DebeziumPropertiesTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023-2023 the original author or authors.
+ * Copyright 2023-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -35,7 +35,6 @@ public class DebeziumPropertiesTests {
 	@Test
 	public void defaultPropertiesTest() {
 		assertThat(this.properties.getPayloadFormat()).isEqualTo(DebeziumFormat.JSON);
-		assertThat(this.properties.getHeaderFormat()).isEqualTo(DebeziumFormat.JSON);
 		assertThat(this.properties.getOffsetCommitPolicy()).isEqualTo(DebeziumOffsetCommitPolicy.DEFAULT);
 		assertThat(this.properties.getProperties()).isNotNull();
 		assertThat(this.properties.getProperties()).isEmpty();