[SUPPORT] missing records when HoodieDeltaStreamer run in continuous mode

[pushpavanthar]

Tips before filing an issue

• Have you gone through our FAQs? Yes

• Join the mailing list to engage in conversations and get faster support at dev-subscribe@hudi.apache.org.

• If you have triaged this as a bug, then file an issue directly. yes

Describe the problem you faced
I've beed running HoodieDeltaStreamer in continuous mode for COW tables in EMR for sometime. To validate the consistency of resultant table, I run a query comparing distinct primary keys created at each hour with the source table. I'm surprised to find some count of unique primary keys in the Hudi table. This implies that HoodieDeltaStreamer in continuous mode for COW tables, the data is not consistent. This issue reappears often in the different streams. Have captured driver logs for investigation.

To Reproduce

Steps to reproduce the behavior:

1. Start EMR cluster with below configs

[{"classification":"hive-env", "properties":{}, "configurations":[{"classification":"export", "properties":{"HADOOP_HEAPSIZE":"5120"}, "configurations":[]}]},{"classification":"presto-connector-hive", "properties":{"hive.parquet.use-column-names":"true", "hive.s3.max-client-retries":"30", "hive.s3select-pushdown.max-connections":"6000", "hive.metastore":"glue", "hive.s3select-pushdown.enabled":"true", "hive.s3-file-system-type":"PRESTO"}, "configurations":[]},{"classification":"yarn-site", "properties":{"yarn.resourcemanager.nodemanager-graceful-decommission-timeout-secs":"600", "yarn.nodemanager.localizer.cache.target-size-mb":"10120", "yarn.nodemanager.localizer.cache.cleanup.interval-ms":"1500000"}, "configurations":[]},{"classification":"capacity-scheduler", "properties":{"yarn.scheduler.capacity.resource-calculator":"org.apache.hadoop.yarn.util.resource.DominantResourceCalculator"}, "configurations":[]},{"classification":"spark-defaults", "properties":{"spark.driver.memory":"4096M", "spark.history.fs.cleaner.maxAge":"1h", "spark.blacklist.decommissioning.timeout":"600s", "spark.port.maxRetries":"32", "spark.history.fs.cleaner.interval":"1h", "spark.history.fs.cleaner.enabled":"true"}, "configurations":[]},{"classification":"emrfs-site", "properties":{"fs.s3.maxRetries":"50"}, "configurations":[]},{"classification":"hive-site", "properties":{"hive.metastore.client.factory.class":"com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory"}, "configurations":[]},{"classification":"spark-hive-site", "properties":{"hive.metastore.client.factory.class":"com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory"}, "configurations":[]}]

2. Submit step to EMR to run deltastreamer as below

spark-submit --master yarn \
--jars /usr/lib/spark/external/lib/spark-avro.jar,s3://lake-bucket/jars/hudi-utilities-bundle_2.12-0.11.1.jar \
--files s3://artifact_bucket/config/hudi/log4j.properties \
--class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer \
--conf spark.executor.cores=3 \
--conf spark.driver.memory=4g \
--conf spark.driver.memoryOverhead=1250m \
--conf spark.executor.memoryOverhead=1250m \
--conf spark.executor.memory=27g \
--conf spark.dynamicAllocation.enabled=true \
--conf spark.dynamicAllocation.initialExecutors=1 \
--conf spark.dynamicAllocation.minExecutors=1 \
--conf spark.dynamicAllocation.maxExecutors=3 \
--conf spark.scheduler.mode=FAIR \
--conf spark.task.maxFailures=5 \
--conf spark.rdd.compress=true \
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
--conf spark.shuffle.service.enabled=true \
--conf spark.sql.hive.convertMetastoreParquet=false \
--conf spark.yarn.max.executor.failures=5 \
--conf spark.sql.catalogImplementation=hive \
--conf spark.driver.cores=3 \
--conf "spark.executor.extraJavaOptions=-Dlog4j.configuration=log4j.properties -XX:+UseG1GC -XX:+PrintFlagsFinal -XX:+PrintReferenceGC -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+PrintAdaptiveSizePolicy -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -XX:ConcGCThreads=6 -XX:ParallelGCThreads=12 -XX:G1HeapRegionSize=33554432 -XX:G1HeapWastePercent=15 -XX:OnOutOfMemoryError='kill -9 %p'" \
--conf "spark.driver.extraJavaOptions=-Dlog4j.configuration=log4j.properties -XX:+UseG1GC -XX:+PrintFlagsFinal -XX:+PrintReferenceGC -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+PrintAdaptiveSizePolicy -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -XX:ConcGCThreads=6 -XX:ParallelGCThreads=12 -XX:G1HeapRegionSize=33554432 -XX:G1HeapWastePercent=15 -XX:OnOutOfMemoryError='kill -9 %p'" \
--conf spark.app.name=cow_workflow_manager_service_kyc \
--conf spark.driver.userClassPathFirst=true \
--deploy-mode cluster s3://lake-bucket/jars/deltastreamer-addons-1.3.jar \
--enable-sync \
--hoodie-conf hoodie.deltastreamer.source.kafka.auto.reset.offsets=earliest \
--hoodie-conf hoodie.parquet.compression.codec=snappy \
--hoodie-conf partition.assignment.strategy=org.apache.kafka.clients.consumer.RangeAssignor \
--hoodie-conf hive.metastore.disallow.incompatible.col.type.changes=false \
--hoodie-conf hoodie.deltastreamer.schemaprovider.spark_avro_post_processor.enable=false \
--hoodie-conf auto.offset.reset=earliest \
--hoodie-conf hoodie.clean.automatic=true \
--hoodie-conf hoodie.clean.async=true \
--hoodie-conf hoodie.clean.max.commits=30 \
--hoodie-conf hoodie.table.services.enabled=true \
--hoodie-conf hoodie.metadata.enable=false \
--table-type COPY_ON_WRITE \
--source-class com.domain.sources.ConfluentAvroKafkaSource \
--schemaprovider-class org.apache.hudi.utilities.schema.NullTargetSchemaRegistryProvider \
--props s3://artifact_bucket/config/kafka/kafka-source.properties \
--source-limit 1000000 \
--hoodie-conf hoodie.deltastreamer.schemaprovider.registry.url=https://schema-registry.in/subjects/workflow_manager_service.public.kyc-value/versions/latest \
--hoodie-conf hoodie.datasource.hive_sync.database=workflow_manager_service \
--hoodie-conf hoodie.datasource.hive_sync.table=kyc \
--hoodie-conf hoodie.datasource.write.recordkey.field=id \
--hoodie-conf hoodie.datasource.write.precombine.field=__lsn \
--hoodie-conf hoodie.deltastreamer.source.kafka.topic=workflow_manager_service.public.kyc \
--hoodie-conf group.id=cds-workflow_manager_service-kyc \
--hoodie-conf hoodie.cleaner.policy=KEEP_LATEST_COMMITS \
--hoodie-conf hoodie.keep.max.commits=6000 \
--hoodie-conf hoodie.keep.min.commits=5000 \
--hoodie-conf hoodie.cleaner.commits.retained=4500 \
--hoodie-conf hoodie.cleaner.parallelism=500 \
--hoodie-conf hoodie.clean.allow.multiple=false \
--hoodie-conf hoodie.cleaner.incremental.mode=true \
--hoodie-conf hoodie.archive.async=true \
--hoodie-conf hoodie.archive.automatic=true \
--hoodie-conf hoodie.archive.merge.files.batch.size=60 \
--hoodie-conf hoodie.commits.archival.batch=30 \
--hoodie-conf hoodie.archive.delete.parallelism=500 \
--hoodie-conf hoodie.archive.merge.enable=true \
--hoodie-conf hoodie.clustering.inline=false \
--hoodie-conf hoodie.index.type=GLOBAL_BLOOM \
--hoodie-conf hoodie.write.markers.type=DIRECT \
--source-ordering-field __lsn \
--target-base-path s3://lake-bucket/raw-data/workflow_manager_service/kyc \
--target-table kyc \
--payload-class com.domain.payload.PostgresSoftDeleteDebeziumAvroPayload \
--hoodie-conf hoodie.bloom.index.update.partition.path=false \
--hoodie-conf hoodie.metrics.on=true \
--hoodie-conf hoodie.metrics.reporter.type=PROMETHEUS_PUSHGATEWAY \
--hoodie-conf hoodie.metrics.pushgateway.host=pushgateway.in \
--hoodie-conf hoodie.metrics.pushgateway.port=443 \
--hoodie-conf hoodie.metrics.pushgateway.delete.on.shutdown=false \
--hoodie-conf hoodie.metrics.pushgateway.job.name=hudi_continuous_workflow_manager_service_kyc_hudi \
--hoodie-conf hoodie.metrics.pushgateway.random.job.name.suffix=false \
--hoodie-conf hoodie.metrics.reporter.metricsname.prefix=hudi \
--hoodie-conf hoodie.datasource.write.partitionpath.field='' \
--hoodie-conf hoodie.datasource.write.keygenerator.class=org.apache.hudi.keygen.NonpartitionedKeyGenerator \
--hoodie-conf hoodie.datasource.hive_sync.partition_extractor_class=org.apache.hudi.hive.NonPartitionedExtractor \
--transformer-class com.domain.transform.DebeziumTransformer \
--hoodie-conf hoodie.deltastreamer.source.kafka.enable.commit.offset=true \
--sync-tool-classes org.apache.hudi.aws.sync.AwsGlueCatalogSyncTool \
--continuous

3. Keep it running for sometime maybe few days and keep comparing with below query. This confirms data missing in Hudi tables compared when compared with raw data(debezium output) retained into s3.

with raw_table as (
  select 'workflow_manager_service_public_kyc' as table_name, date_format(from_iso8601_timestamp(created_at), '%Y%m%d %H') as created_at,
    count(distinct id) as raw_count
  FROM raw_cdc_db.raw_workflow_manager_service_public_kyc
  where date_format(from_iso8601_timestamp(created_at), '%Y%m%d') >= date_format((NOW() - INTERVAL '3' DAY), '%Y%m%d')
    and date_format(from_iso8601_timestamp(created_at), '%Y%m%d %H') <= date_format((NOW() - INTERVAL '1' HOUR), '%Y%m%d %H')
    and dt >= date_format((NOW() - INTERVAL '7' DAY), '%Y%m%d')
  group by 1, 2
),
hudi_table as (
  select 'workflow_manager_service_public_kyc' as table_name, date_format(from_iso8601_timestamp(created_at), '%Y%m%d %H') as p_date,
    count(1) as hudi_count
  from workflow_manager_service.kyc where date_format(from_iso8601_timestamp(created_at), '%Y%m%d') >= date_format((NOW() - INTERVAL '3' DAY), '%Y%m%d')
  and date_format(from_iso8601_timestamp(created_at), '%Y%m%d %H') <= date_format((NOW() - INTERVAL '1' HOUR), '%Y%m%d %H')
  group by 1, 2
),
union_table as (
  select coalesce(l.table_name, r.table_name) as table_name, 
    coalesce(l.created_at, r.p_date) as create_at_dt,
    l.raw_count,
    r.hudi_count
  from raw_table l
    FULL OUTER JOIN hudi_table r on (l.created_at = r.p_date and l.table_name = r.table_name)
)
select *
from union_table
where raw_count is NULL or hudi_count is NULL
  or raw_count != hudi_count

Expected behavior
Hudi table should have exact number of records for each hour when compared with raw events. Hudi having less records compared to Raw table implies that data is missing from Hudi. Can someone help me where to start with to debug this issue?

Environment Description

• Hudi version : 0.11.1

• Spark version : 3.1.1

• Hive version : 3.1.2

• Hadoop version : Amazon 3.2.1

• Storage (HDFS/S3/GCS..) : S3

• Running on Docker? (yes/no) : No

Additional context
Driver logs are retained if it can be of some help for analysis.

[codope]

@pushpavanthar Thanks for sharing the details. Few notes on the validation:

1. The filter for raw_table and hudi_table is not the same.
2. There could be a mismatch on hourly basis because dynamic allocation is enabled and executors could get lost resulting in failure of a deltastreamer round. However, the checkpoint would not be updated so the records will get picked up in the subsequent deltastreamer round. But in this case, it is possible to see less records in hudi table than the raw table. It depends on when the validation query ran and how far behind the source table is the deltastreamer lagging (essentially due to lag between processing time and event time).
3. For the hudi_table, it would be helpful to collect _hoodie_commit_time values for a particular created_at_dt value. Then, we can look into the timeline around that commit time for further debugging.
4. I understand you are running in continuous mode and source data is being continuously updated. But, let's say you know for sure that yesterday's data has been processed, then can you do a validation of record count across both tables with a simple filter on created_at like select count(*) from table where created_at = '<yesterday date>'.

Few notes on the configuration:

1. Record key is id and precombine field is _lsn. I am assuming all records in the source table have unique id, otherwise Hudi will dedup the records based on _lsn and there could be lesser number of absolute records in Hudi table.
2. I see that schema registry is being used as the schema provider but --schemaprovider-class is set to NullTargetSchemaRegistryProvider. Is it because there is some additional transformation on the source data before ingesting into Hudi table and your requirement is the Hudi be able to infer the target schema? If so, can you also share what the transformer is doing in this case.
3. Is it possible disable dynamic allocation for a few hours or a day and then validate records processed for that day?

[pushpavanthar]

Thanks for looking into this issue @codope. Below is a brief explanation to points you mentioned.

1. The data in raw table is written by s3-sink connector which roll files every 15 mins and partitioned by date derived from kafka metadata timestamp. I'm checking for count of unique primary keys per created_at hour (at least number of create records should match) for the last 3 days by excluding the current hour (to avoid inconsistencies in the current hour due to difference in nature of pipeline). Still I have a buffer of scanning 7+ days to account outliers to compare data of last 3 days. FYI my hudi tables are not partitioned.
2. For this pipeline we have provided sufficient resources and are constantly monitoring for lag. haven't noticed anything strange wrt application and cluster. Similar to your observation, I'm suspecting on hoodie.deltastreamer.source.kafka.enable.commit.offset: true config, which lets consumer groups in kafka to manage offsets. There might be a situation where consumer offsets are committed to kafka and some failure in the cycle might have triggered rollback. Next cycle of deltasync would pick from next set of offsets, hence miss entire batch of old records.
   I'll try running few pipelines by disabling this config.
3. For now i've replayed the events to correct the inconsistencies since it impacts our reports. Have seen similar issues in the past on other tables. Will do this analysis when I come across this issue again.
4. The count of unique records in entire table is lower in Hudi table compared to raw table. Hence to dig deep, I ran query for hourly comparison.

for the notes on configurations

1. Verified that all records are having unique ids. The hourly distinct count of unique ids on raw tables are matching with the source db but doesn't match with hudi table.
2. We are doing some transformation to drop __op and __source_ts_ms and explicitly set _hoodie_is_deleted with false make sure we retain deleted records also.
3. Will try out disabling dynamic allocation.

[pushpavanthar]

disabling hoodie.deltastreamer.source.kafka.enable.commit.offset made no difference.
The problem is still there. Another strange behaviour is, to rectify the table, when I run deltastreamer with --checkpoint pointing to earliest offset, the first commit will have all records, however, from next commit onwards the records will go missing. Confirmed this behaviour by running queries for missing record keys using as.of.instant feature.
To explain the problem better, below is the timeline of commits occurred

1. 20230130083344298.commit //commit till the job ran last>
2. 20230130084045370.commit //restarted Deltastreamer with checkpoint>
3. 20230130084529555.commit //Consecutive commits from here onwards>
4. 20230130084944755.commit

All the above commits are updating fileId c77a6ee1-ab98-47c9-9cdc-fd17b167bf8a-0. If My understanding is right, writestats should contain commit instant of latest commit which had updated on this fileId. However, in 20230130084529555.commit it shows "prevCommit" : "20230130083344298" in its writestats. This clearly shows, the consecutive deltasync runs didn't pick fileslices from latest commit which skips some updates.

  "partitionToWriteStats" : {
    "" : [ {
      "fileId" : "c77a6ee1-ab98-47c9-9cdc-fd17b167bf8a-0",
      "path" : "c77a6ee1-ab98-47c9-9cdc-fd17b167bf8a-0_0-64-3632_20230130084529555.parquet",
      "prevCommit" : "20230130083344298",
      "numWrites" : 235286,
      "numDeletes" : 0,
      "numUpdateWrites" : 9,
      "numInserts" : 49,
      "totalWriteBytes" : 72504131,
      "totalWriteErrors" : 0,
      "tempPath" : null,
      "partitionPath" : "",
      "totalLogRecords" : 0,
      "totalLogFilesCompacted" : 0,
      "totalLogSizeCompacted" : 0,
      "totalUpdatedRecordsCompacted" : 0,
      "totalLogBlocks" : 0,
      "totalCorruptLogBlock" : 0,
      "totalRollbackBlocks" : 0,
      "fileSizeInBytes" : 72504131,
      "minEventTime" : null,
      "maxEventTime" : null
    } ]
  },

[pratyakshsharma]

@codope Do you have cycles to check the latest comments here?

[codope]

@pushpavanthar Ok that's some useful information. I see that there is a rollback commit in between. What could have possible triggered that? Is it possible to check from logs? Also, if you can provide a zip of the timeline, it could help us to debug the issue.

[pushpavanthar]

@codope I have identified root cause and fix for the same, can you help me create a jira and assign to me. I'll update the approach to fix this issue.

[codope]

@pushpavanthar That's good news! You can also outline the cause and fix here. JIRA: HUDI-5686
Please tag me in the PR once it is ready for review.

[danny0405]

It's a known critical bug for 0.11.x release, I have put a fix for 0.12.0: #6179