deepakpanda93 commented on code in PR #19597: URL: https://github.com/apache/hudi/pull/19597#discussion_r3773434200
########## website/docs/hoodie_streaming_ingestion.md: ########## @@ -628,6 +628,86 @@ Using `org.apache.hudi.utilities.sources.SqlFileBasedSource` allows setting the table. SQL file path should be configured using this hoodie config: `hoodie.streamer.source.sql.file = 'hdfs://xxx/source.sql'` +#### Debezium + +Hudi Streamer can keep a Hudi table in sync with an upstream database by ingesting change data capture (CDC) events +produced by [Debezium](https://debezium.io/). Debezium publishes each change as an Avro message on a Kafka topic and +registers the schema with a Confluent schema registry. The Debezium sources read that topic, flatten the nested Debezium +change envelope into ordinary table columns, and apply the resulting inserts, updates and deletes to the target table. + +There is one source and one matching payload class per database: + +| Database | Source class | Payload class | +|------------|---------------------------------------------------------------------|---------------------------------------------------------------------| +| PostgreSQL | `org.apache.hudi.utilities.sources.debezium.PostgresDebeziumSource` | `org.apache.hudi.common.model.debezium.PostgresDebeziumAvroPayload` | +| MySQL | `org.apache.hudi.utilities.sources.debezium.MysqlDebeziumSource` | `org.apache.hudi.common.model.debezium.MySqlDebeziumAvroPayload` | + +Note that the two halves spell MySQL differently: the source is `Mysql...` while the payload is `MySql...`. + +Both sources read Avro and require a schema registry, so set `--schemaprovider-class` to +`org.apache.hudi.utilities.schema.SchemaRegistryProvider` and point `hoodie.streamer.schemaprovider.registry.url` at the +subject for the topic. The Kafka value deserializer already defaults to +`io.confluent.kafka.serializers.KafkaAvroDeserializer`, so `hoodie.streamer.source.kafka.value.deserializer.class` only +needs setting in order to override it. + +A property file for a PostgreSQL table: + +```properties +hoodie.streamer.source.kafka.topic=postgres.public.customers +hoodie.streamer.schemaprovider.registry.url=http://localhost:8081/subjects/postgres.public.customers-value/versions/latest +bootstrap.servers=localhost:9092 Review Comment: You're right, and this one was a genuine defect rather than a polish item — the example as written would have failed at startup. Fixed in 75d01d2b. I verified the mechanism rather than taking the report on trust, and every step of it holds. `KafkaOffsetGen.excludeHoodieConfigs` builds the consumer params by dropping every `hoodie.*` key, with exactly one exception: ```java return !prop.toString().startsWith("hoodie.") // We need to pass some properties to kafka client so that KafkaAvroSchemaDeserializer can use it || prop.toString().startsWith(AvroKafkaSource.KAFKA_AVRO_VALUE_DESERIALIZER_PROPERTY_PREFIX) ``` and that prefix is `hoodie.streamer.source.kafka.value.deserializer.`, which `hoodie.streamer.schemaprovider.registry.url` does not match. So the schema provider's URL is stripped and never reaches the deserializer, exactly as you describe. I also checked the other escape route, since it was the one thing that could have made the report wrong: `DebeziumSource` calls `KafkaSourceUtil.configureSchemaDeserializer` only under `deserializerClassName.equals(KafkaAvroSchemaDeserializer.class.getName())`, and this example keeps the Confluent default, so that path does not run either. The corroboration you pointed at is the part that settles it. `hudi-utilities/src/test/resources/streamer-config/kafka-source.properties` sets both keys: ```properties hoodie.streamer.schemaprovider.registry.url=http://localhost:8081/subjects/impressions-value/versions/latest ... schema.registry.url=http://localhost:8081 ``` That is the same file the Kafka example further up this page passes to `--props`, so the page was already implicitly relying on the two-key pattern while my new example dropped one of them. Checked across the version range too, since this PR touches six copies: `excludeHoodieConfigs` strips `hoodie.*` on all five 1.x tags, and all five ship a `kafka-source.properties` that sets `schema.registry.url` explicitly. So the fix is correct everywhere and needed no per-version variation. The fix adds `schema.registry.url=http://localhost:8081` to the property file, and says in the prose *why* the registry appears twice, so the second entry doesn't read as a copy-paste slip: > The registry has to be given to the Kafka consumer a second time, as a plain `schema.registry.url`, because Hudi drops every `hoodie.*` property before it constructs the consumer and the schema provider's URL therefore never reaches the deserializer. Build passes with the warning block still byte-identical to the baseline at `d11a5b0adee4`; both the property and the prose render on `next` and all five 1.x copies. Worth recording how this got in, since it is the more useful lesson: I had `schema.registry.url` in the draft and removed it late, on the reasoning that I had not independently confirmed it and should keep the example to keys I had verified. Trimming an unverified line turned a working example into a broken one — for a runnable recipe, an omission is not the safe direction. This is also the limitation I flagged in the PR description: the example was verified by reading the code paths it invokes, not by executing it, and executing it is precisely what would have caught this. Still happy to stand up a live Debezium → Kafka → registry → Hudi run before merge if that would help. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
