sydneyhoran opened a new issue, #8372:
URL: https://github.com/apache/hudi/issues/8372
**Describe the problem you faced**
Using Deltastreamer to ingest Avro messages from Kafka topics
(PostgresDebeziumAvroPayload) into S3 Hudi tables. When using
CustomKeyGenerator for multiple RecordKeys and timebased PartitionPath,
Deltastreamer throws an error on first run if the partitionPath doesn't include
the datatype.
Configs:
```
hoodie.datasource.write.recordkey.field=id,_changed_at,_change_type
hoodie.datasource.write.partitionpath.field=inserted_at
hoodie.datasource.write.keygenerator.class=org.apache.hudi.keygen.CustomKeyGenerator
hoodie.deltastreamer.keygen.timebased.timestamp.type=EPOCHMILLISECONDS
hoodie.deltastreamer.keygen.timebased.output.dateformat=yyyy/MM/dd
hoodie.datasource.write.precombine.field=updated_at
```
Error:
```
Caused by: org.apache.hudi.exception.HoodieKeyException: Unable to find
field names for partition path in proper format
```
We found that the first job must be run with the datatype (i.e.
`hoodie.datasource.write.partitionpath.field=inserted_at:TIMESTAMP`), and Hudi
automatically sets `hoodie.datasource.write.partitionpath.field` with the
extracted partition path as `inserted_at` (without the :TIMESTAMP). However, on
the following run, when we pass `inserted_at:TIMESTAMP`, the job fails due to
config conflict.
```
Exception in thread "main" org.apache.hudi.exception.HoodieException: Config
conflict(key current value existing value):
PartitionPath: inserted_at:TIMESTAMP inserted_at
```
The :TIMESTAMP must be removed for subsequent runs, but appears to be
required for the first run. The expected behavior is to allow configs be the
same for first run vs all future runs.
Is there another config we can set that would be compatible for first run as
well as future runs?
**To Reproduce**
Steps to reproduce the behavior:
1. Run Deltastreamer for Debezium Avro payload messages from Kafka topic
2. Set configs to use CustomKeyGenerator, multiple record keys,
EPOCHMILLISECONDS formatted partition path (don't set :TIMESTAMP) for the first
run
3. HoodieKeyException is produced `Unable to find field names for partition
path in proper format`
4. Change config to
`hoodie.datasource.write.partitionpath.field=inserted_at:TIMESTAMP`
5. Job runs as expected
6. Run the job again with the same config ^ (:TIMESTAMP)
7. HoodieException `Config conflict` is produced
8. Remove :TIMESTAMP from partitionpath config
9. Job runs as expected
**Expected behavior**
Deltastreamer should recognize the PartitionPath column on the first run,
and configs should be the same for all future runs. If datatype needs to be
specified, it should be put into the `.hoodie/hoodie.properties` file as
inserted_at:TIMESTAMP so there is no config conflict later.
**Environment Description**
* Hudi version : 0.14.0-SNAPSHOT
* Spark version : 3.1
* Hive version : N/A
* Hadoop version : N/A
* Storage (HDFS/S3/GCS..) : S3
* Running on Docker? (yes/no) : both
**Additional context**
Add any other context about the problem here.
**Stacktrace**
**For first run without specifying :TIMESTAMP:**
```
Caused by: org.apache.hudi.exception.HoodieKeyException: Unable to find
field names for partition path in proper format
at
org.apache.hudi.keygen.CustomAvroKeyGenerator.getPartitionPath(CustomAvroKeyGenerator.java:78)
at
org.apache.hudi.keygen.CustomKeyGenerator.getPartitionPath(CustomKeyGenerator.java:83)
at
org.apache.hudi.keygen.BaseKeyGenerator.getKey(BaseKeyGenerator.java:69)
at
org.apache.hudi.utilities.deltastreamer.DeltaSync.lambda$fetchFromSource$ddc224fb$1(DeltaSync.java:618)
at
org.apache.spark.api.java.JavaPairRDD$.$anonfun$toScalaFunction$1(JavaPairRDD.scala:1070)
at scala.collection.Iterator$$anon$10.next(Iterator.scala:461)
at scala.collection.Iterator$SliceIterator.next(Iterator.scala:273)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
at scala.collection.generic.Growable.$plus$plus$eq(Growable.scala:62)
at scala.collection.generic.Growable.$plus$plus$eq$(Growable.scala:53)
at
scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:105)
at
scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:49)
at scala.collection.TraversableOnce.to(TraversableOnce.scala:366)
at scala.collection.TraversableOnce.to$(TraversableOnce.scala:364)
at scala.collection.AbstractIterator.to(Iterator.scala:1431)
at scala.collection.TraversableOnce.toBuffer(TraversableOnce.scala:358)
at scala.collection.TraversableOnce.toBuffer$(TraversableOnce.scala:358)
at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1431)
at scala.collection.TraversableOnce.toArray(TraversableOnce.scala:345)
at scala.collection.TraversableOnce.toArray$(TraversableOnce.scala:339)
at scala.collection.AbstractIterator.toArray(Iterator.scala:1431)
at org.apache.spark.rdd.RDD.$anonfun$take$2(RDD.scala:1449)
at
org.apache.spark.SparkContext.$anonfun$runJob$5(SparkContext.scala:2244)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:131)
at
org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:505)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1439)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:508)
... 3 more
```
**For subsequent runs keeping :TIMESTAMP in the config:**
```
KeyGenerator: org.apache.hudi.keygen.CustomKeyGenerator
org.apache.hudi.keygen.TimestampBasedKeyGenerator
PartitionPath: inserted_at:TIMESTAMP inserted_at
at
org.apache.hudi.HoodieWriterUtils$.validateTableConfig(HoodieWriterUtils.scala:181)
at
org.apache.hudi.HoodieWriterUtils$.validateTableConfig(HoodieWriterUtils.scala:131)
at
org.apache.hudi.HoodieWriterUtils.validateTableConfig(HoodieWriterUtils.scala)
at
org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer$DeltaSyncService.<init>(HoodieDeltaStreamer.java:647)
at
org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer.<init>(HoodieDeltaStreamer.java:156)
at
org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer.<init>(HoodieDeltaStreamer.java:129)
at
org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer.main(HoodieDeltaStreamer.java:559)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at
org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)
at
org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:951)
at
org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:180)
at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:203)
at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:90)
at
org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:1039)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:1048)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
```
--
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]
