DBZ-1003 Initialize database history

debezium-core/src/main/java/io/debezium/relational/HistorizedRelationalDatabaseSchema.java

@@ -51,6 +51,17 @@ public void close() {
         databaseHistory.stop();
     }
 
+    /**
+     * Configures a storage used to store history, e.g. in Kafka case it creates topic with
+     * required parameters.
+     */
+    @Override
+    public void initializeStorage() {
+        if (!databaseHistory.exists()) {
+            databaseHistory.initializeStorage();
+        }
+    }
+
     /**
      * Returns a new instance of the {@link DdlParser} to be used when recovering the schema from a previously persisted
      * history.

debezium-core/src/main/java/io/debezium/schema/HistorizedDatabaseSchema.java

@@ -21,4 +21,6 @@ public interface HistorizedDatabaseSchema<I extends DataCollectionId> extends Da
     void applySchemaChange(SchemaChangeEvent schemaChange);
 
     void recover(OffsetContext offset);
+
+    void initializeStorage();
 }