Wire DebeziumSDFDatabaseHistory default to schema.history.internal

The default schema history implementation was registered under the
pre-Debezium-2.0 key database.history, which Debezium 3.x ignores.
As a result DebeziumSDFDatabaseHistory was unreachable unless users set
schema.history.internal explicitly and historized connectors silently
fell back to KafkaSchemaHistory pointing at localhost:9092. Registering
the default under schema.history.internal makes the Beam native history
implementation the default again as originally intended. Users who set
schema.history.internal themselves are unaffected.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: tkaymak

sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumIO.java

@@ -74,7 +74,7 @@
  *             .withConnectorClass(MySqlConnector.class)
  *             .withConnectionProperty("database.server.id", "184054")
  *             .withConnectionProperty("database.server.name", "serverid")
- *             .withConnectionProperty("database.history", DebeziumSDFDatabaseHistory.class.getName())
+ *             .withConnectionProperty("schema.history.internal", DebeziumSDFDatabaseHistory.class.getName())
  *             .withConnectionProperty("include.schema.changes", "false");
  *
  *      PipelineOptions options = PipelineOptionsFactory.create();
@@ -640,10 +640,10 @@ public Map<String, String> getConfigurationMap() {
         configuration.computeIfAbsent(entry.getKey(), k -> entry.getValue());
       }
 
-      // Set default Database History impl. if not provided implementation and Kafka topic prefix,
-      // if not provided
+      // Set default schema history impl. if not provided implementation and Kafka topic prefix,
+      // if not provided. Before Debezium 2.0 this key was named "database.history".
       configuration.computeIfAbsent(
-          "database.history",
+          "schema.history.internal",
           k -> KafkaSourceConsumerFn.DebeziumSDFDatabaseHistory.class.getName());
       configuration.computeIfAbsent("topic.prefix", k -> "beam-debezium-connector");
       configuration.computeIfAbsent(

sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOTest.java

@@ -52,7 +52,8 @@ public class DebeziumIOTest implements Serializable {
           .withConnectionProperty("database.server.id", "184054")
           .withConnectionProperty("database.server.name", "dbserver1")
           .withConnectionProperty(
-              "database.history", KafkaSourceConsumerFn.DebeziumSDFDatabaseHistory.class.getName())
+              "schema.history.internal",
+              KafkaSourceConsumerFn.DebeziumSDFDatabaseHistory.class.getName())
           .withConnectionProperty("include.schema.changes", "false");
 
   @Test