[pulsar-io] pass the pulsar service url to debezium source for histor…

…y database (apache#11251)

### Motivation

The Debezium requires pulsar a service URL for history database usage. 

In apache#11056 , the `service.url` field from `PulsarKafkaWorkerConfig` is no longer available. And the value is also deleted from multiple yaml config files in this [commit](apache@3ce24c9). This causes the integration test for Debezium connector to fail.

Based on the Debezium [paradigm](https://debezium.io/documentation/reference/1.5/connectors/mysql.html#debezium-mysql-connector-database-history-configuration-properties), all configurations should be passed as strings. There's no easy way to inject a PulsarClient via configuration.

We need to ask user to provide the pulsar url explicitly and probably auth info also.


### Modifications

1. Make the `database.history.pulsar.service.url` field required
2. Add the config value back to example yaml files
3. Update the integration test config

### Verifying this change

- [ ] Make sure that the change passes the CI checks.

pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/DebeziumSource.java

@@ -78,6 +78,12 @@ public void open(Map<String, Object> config, SourceContext sourceContext) throws
         // database.history : implementation class for database history.
         setConfigIfNull(config, HistorizedRelationalDatabaseConnectorConfig.DATABASE_HISTORY.name(), DEFAULT_HISTORY);
 
+        // database.history.pulsar.service.url
+        String pulsarUrl = (String) config.get(PulsarDatabaseHistory.SERVICE_URL.name());
+        if (StringUtils.isEmpty(pulsarUrl)) {
+            throw new IllegalArgumentException("Pulsar service URL for History Database not provided.");
+        }
+
         String topicNamespace = topicNamespace(sourceContext);
         // topic.namespace
         setConfigIfNull(config, PulsarKafkaWorkerConfig.TOPIC_NAMESPACE_CONFIG, topicNamespace);

pulsar-io/debezium/mongodb/src/main/resources/debezium-mongodb-source-config.yaml

@@ -33,3 +33,5 @@ configs:
   mongodb.password: "dbz"
   mongodb.task.id: "1"
   database.whitelist: "inventory"
+
+  database.history.pulsar.service.url: "pulsar://127.0.0.1:6650"

pulsar-io/debezium/mysql/src/main/resources/debezium-mysql-source-config.yaml

@@ -36,6 +36,7 @@ configs:
   database.whitelist: "inventory"
 
   database.history.pulsar.topic: "mysql-history-topic"
+  database.history.pulsar.service.url: "pulsar://127.0.0.1:6650"
   offset.storage.topic: "mysql-offset-topic"
 
 

pulsar-io/debezium/postgres/src/main/resources/debezium-postgres-source-config.yaml

@@ -35,5 +35,4 @@ configs:
   database.server.name: "dbserver1"
   schema.whitelist: "inventory"
 
-
-
+  database.history.pulsar.service.url: "pulsar://127.0.0.1:6650"

tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMongoDbSourceTester.java

@@ -50,7 +50,7 @@ public DebeziumMongoDbSourceTester(PulsarCluster cluster) {
         sourceConfig.put("mongodb.password", "dbz");
         sourceConfig.put("mongodb.task.id","1");
         sourceConfig.put("database.whitelist", "inventory");
-        sourceConfig.put("pulsar.service.url", pulsarServiceUrl);
+        sourceConfig.put("database.history.pulsar.service.url", pulsarServiceUrl);
         sourceConfig.put("topic.namespace", "debezium/mongodb");
     }
 

tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMySqlSourceTester.java

@@ -61,7 +61,7 @@ public DebeziumMySqlSourceTester(PulsarCluster cluster, String converterClassNam
         sourceConfig.put("database.server.id", "184054");
         sourceConfig.put("database.server.name", "dbserver1");
         sourceConfig.put("database.whitelist", "inventory");
-        sourceConfig.put("pulsar.service.url", pulsarServiceUrl);
+        sourceConfig.put("database.history.pulsar.service.url", pulsarServiceUrl);
         sourceConfig.put("key.converter", converterClassName);
         sourceConfig.put("value.converter", converterClassName);
         sourceConfig.put("topic.namespace", "debezium/mysql-" +

tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumPostgreSqlSourceTester.java

@@ -63,7 +63,7 @@ public DebeziumPostgreSqlSourceTester(PulsarCluster cluster) {
         sourceConfig.put("database.dbname", "postgres");
         sourceConfig.put("schema.whitelist", "inventory");
         sourceConfig.put("table.blacklist", "inventory.spatial_ref_sys,inventory.geom");
-        sourceConfig.put("pulsar.service.url", pulsarServiceUrl);
+        sourceConfig.put("database.history.pulsar.service.url", pulsarServiceUrl);
         sourceConfig.put("topic.namespace", "debezium/postgresql");
     }