shangxinli commented on code in PR #20445:
URL: https://github.com/apache/hudi/pull/20445#discussion_r4484661063
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala:
##########
@@ -315,18 +333,20 @@ class HoodieStreamingSink(sqlContext: SQLContext,
}
/**
- * Same as HoodieStreamer on a schema change (StreamSync#reInitWriteClient
and
- * HoodieStreamer#onInitializingWriteClient): the async table services
switch to the new client for their next
- * runs, and the previous client is closed right away.
+ * Closes the previous client right away, then switches the async table
services to the new client for their next
+ * runs; the switch does not wait for a running compaction or clustering. A
run still using the previous client can
+ * fail once it is closed: the previous client is marked as retired first,
so that its service retries such a
+ * failure with the new client instead of stopping.
*/
private def retireWriteClient(previousClient: SparkRDDWriteClient[_],
newClient: SparkRDDWriteClient[_]): Unit = {
Review Comment:
nit: moving closeQuietly ahead of the compactor/clustering service handoff
means a slow or blocking close() now delays those updates, where before the
handoff happened first.
##########
hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/async/SparkStreamingAsyncClusteringService.java:
##########
@@ -31,8 +33,27 @@ public class SparkStreamingAsyncClusteringService extends
AsyncClusteringService
private static final long serialVersionUID = 1L;
+ // whether a write client was retired by the sink, i.e. closed by the sink
when it replaced its client
+ private final transient Predicate<BaseHoodieWriteClient> isRetiredClient;
+
public SparkStreamingAsyncClusteringService(HoodieEngineContext context,
BaseHoodieWriteClient writeClient) {
+ this(context, writeClient, c -> false);
+ }
+
+ public SparkStreamingAsyncClusteringService(HoodieEngineContext context,
BaseHoodieWriteClient writeClient,
+ Predicate<BaseHoodieWriteClient> isRetiredClient) {
super(context, writeClient, true);
+ this.isRetiredClient = isRetiredClient;
+ }
+
+ /**
+ * A clustering run that failed while using a write client the sink has
since retired is retryable: the sink closes
+ * the previous client when it re-creates its client (e.g. on a schema
change), which can fail a run still using it.
+ * Any other failure is not retryable and stops the service, which fails the
next micro-batch with that failure.
+ */
+ @Override
+ protected boolean isRetryableFailure(BaseHoodieWriteClient runClient,
Exception failure) {
+ return isRetiredClient.test(runClient);
Review Comment:
minor: this returns isRetiredClient.test(runClient) without looking at the
exception, so any failure after a client is retired gets treated as retryable
even if it's an unrelated real error.
--
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]
