debezium/dbz#1469 introduce DebeziumFactory for db2

Signed-off-by: kmos <kmos@commonhaus.dev>

Author: kmos

quarkus-debezium-parent/quarkus-debezium-db2-parent/runtime/src/main/java/io/quarkus/debezium/engine/Db2EngineProducer.java

@@ -22,19 +22,16 @@
 import io.quarkus.datasource.common.runtime.DatabaseKind;
 import io.quarkus.debezium.agroal.engine.AgroalParser;
 import io.quarkus.debezium.configuration.DebeziumConfigurationEngineParser;
-import io.quarkus.debezium.engine.capture.consumer.SourceRecordConsumerHandler;
 
 public class Db2EngineProducer implements ConnectorProducer {
     public static final Connector DB2 = new Connector(Db2Connector.class.getName());
 
-    private final StateHandler stateHandler;
-    private final SourceRecordConsumerHandler sourceRecordConsumerHandler;
     private final AgroalParser agroalParser;
+    private final DebeziumFactory debeziumFactory;
 
-    public Db2EngineProducer(StateHandler stateHandler, SourceRecordConsumerHandler sourceRecordConsumerHandler, AgroalParser agroalParser) {
-        this.stateHandler = stateHandler;
-        this.sourceRecordConsumerHandler = sourceRecordConsumerHandler;
+    public Db2EngineProducer(AgroalParser agroalParser, DebeziumFactory debeziumFactory) {
         this.agroalParser = agroalParser;
+        this.debeziumFactory = debeziumFactory;
     }
 
     @Override
@@ -46,18 +43,11 @@ public DebeziumConnectorRegistry engine(DebeziumEngineConfiguration debeziumEngi
             private final Map<String, Debezium> engines = multiEngineConfigurations
                     .stream()
                     .map(engine -> {
-                        EngineManifest engineManifest = new EngineManifest(engine.engineId());
-
-                        Map<String, String> debeziumConfiguration = engine.configuration();
-
                         // remove unnecessary configuration for sqlserver
-                        debeziumConfiguration.remove(DATABASE_CONFIG_PREFIX + JdbcConfiguration.DATABASE.name());
+                        engine.configuration()
+                                .remove(DATABASE_CONFIG_PREFIX + JdbcConfiguration.DATABASE.name());
 
-                        return Map.entry(engine.engineId(), new SourceRecordDebezium(
-                                engine.configuration(),
-                                stateHandler,
-                                DB2,
-                                sourceRecordConsumerHandler.get(engineManifest), engineManifest));
+                        return Map.entry(engine.engineId(), debeziumFactory.get(DB2, engine));
                     })
                     .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));