hudi-agent commented on code in PR #19811: URL: https://github.com/apache/hudi/pull/19811#discussion_r3953488205
########## hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/ContinuousTestSource.java: ########## @@ -0,0 +1,159 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hudi.utilities.sources; + +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.table.checkpoint.Checkpoint; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.common.util.collection.Pair; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.utilities.schema.SchemaProvider; + +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; + +import java.util.concurrent.BrokenBarrierException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * A parquet test source that proves continuous-mode multi-table syncs run in parallel. + * + * <p>Every table waits at a shared barrier before producing data, so a sequential implementation would block the first + * table forever and time out. Only concurrent syncs let all tables pass the barrier. + */ +public class ContinuousTestSource extends ParquetDFSSource { + + // When set on a table's properties, that table fails right after passing the barrier, i.e. once all tables started. + public static final String FAIL_AFTER_BARRIER = "hoodie.test.continuous.source.fail.after.barrier"; + + // When set on a table's properties, that table blocks after the barrier until fail fast interrupts it. + public static final String BLOCK_UNTIL_INTERRUPTED = "hoodie.test.continuous.source.block.until.interrupted"; + + // When set on a table's properties, that table fails only after a sibling has finished its sync. + public static final String FAIL_AFTER_SIBLING_COMPLETES = "hoodie.test.continuous.source.fail.after.sibling.completes"; + + private static final long BARRIER_TIMEOUT_SECONDS = 60; + // Time for a finished sibling's future to complete before the waiting table fails. Only widens the gap between + // the two completions, so a longer settle can never hide a regression. + private static final long SIBLING_SETTLE_MILLIS = 2000; + + private static volatile CyclicBarrier startBarrier = new CyclicBarrier(1); + // Counted down by a blocking table once it observes the fail-fast interrupt, so a test can assert it was torn down. + private static volatile CountDownLatch blockedTableInterrupted = new CountDownLatch(1); + // Counted down when a table releases its source, which its ingestion service does as the sync ends. + private static volatile CountDownLatch tableCompleted = new CountDownLatch(1); + + private final boolean failAfterBarrier; + private final boolean blockUntilInterrupted; + private final boolean failAfterSiblingCompletes; + private final AtomicBoolean barrierPassed = new AtomicBoolean(false); + + public ContinuousTestSource(TypedProperties props, JavaSparkContext sparkContext, SparkSession sparkSession, + SchemaProvider schemaProvider) { + super(props, sparkContext, sparkSession, schemaProvider); + this.failAfterBarrier = props.getBoolean(FAIL_AFTER_BARRIER, false); + this.blockUntilInterrupted = props.getBoolean(BLOCK_UNTIL_INTERRUPTED, false); + this.failAfterSiblingCompletes = props.getBoolean(FAIL_AFTER_SIBLING_COMPLETES, false); + } + + // Resets the shared barrier and latch used to coordinate tables. Call before each sync. + public static void resetBarrier(int numTables) { + startBarrier = new CyclicBarrier(numTables); + blockedTableInterrupted = new CountDownLatch(1); + tableCompleted = new CountDownLatch(1); + } + + // Whether a blocking table has already observed the fail-fast interrupt. + public static boolean wasBlockedTableInterrupted() { + return blockedTableInterrupted.getCount() == 0; + } + + @Override + public Pair<Option<Dataset<Row>>, Checkpoint> fetchNextBatch(Option<Checkpoint> lastCheckpoint, long sourceLimit) { + // Only rendezvous once, on the first fetch, so that later empty polls do not block termination. + if (barrierPassed.compareAndSet(false, true)) { + awaitBarrier(); + if (failAfterBarrier) { + throw new HoodieException("Simulated table sync failure after all tables started"); + } + if (blockUntilInterrupted) { + blockUntilFailFastInterrupts(); + } + if (failAfterSiblingCompletes) { + awaitSiblingCompleted(); + throw new HoodieException("Simulated table sync failure after a sibling table finished normally"); + } + } + return super.fetchNextBatch(lastCheckpoint, sourceLimit); + } + + @Override + public void releaseResources() { + // Counting down twice is harmless; only the first table to finish matters. + tableCompleted.countDown(); Review Comment: 🤖 I don't think this fires when the sibling's sync ends — `StreamSync.syncOnce()` calls `source.releaseResources()` in its `finally`, so `tableCompleted` counts down after table 1's *first* round. With `NoNewDataTerminationStrategy` defaulting to 3 empty rounds, table 1 still has three more rounds plus `close()` to get through inside the 2s settle; if it doesn't, table 2's failure interrupts table 1 mid-sync, it lands in `failedTables`, and `assertEquals(1, failedTables.size())` flakes. Could the test instead gate on a real completion signal, e.g. run `sync()` on a background thread and count the latch down once `getSuccessTables()` contains table 1 (it's a `ConcurrentHashMap` key set, so it's safely readable)? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> -- 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]
