1996fanrui commented on code in PR #28305:
URL: https://github.com/apache/flink/pull/28305#discussion_r3638472694
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -1188,7 +1196,9 @@ public boolean hasMail() {
private boolean taskIsAvailable() {
return recordWriter.isAvailable()
&& (changelogWriterAvailabilityProvider == null
- || changelogWriterAvailabilityProvider.isAvailable());
+ || changelogWriterAvailabilityProvider.isAvailable())
+ && (operatorChainAvailabilityProvider == null
+ || operatorChainAvailabilityProvider.isAvailable());
Review Comment:
Could the folded provider replace `recordWriter` in the gate instead of
being ANDed — i.e. drop `recordWriter.isAvailable()` here, and in
`processInput` merge the new 4th suspend branch into the 1st
soft-backpressure branch (same metric)? Regular chains would be unchanged:
the folded provider equals the record writers there — in fact it's currently
non-null for every chain with network outputs, so they check the same writers
twice.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java:
##########
@@ -232,6 +239,22 @@ public OperatorChain(
mainOperatorAndTimeService.f1,
true);
+ // Soft-backpressure head absorbs downstream availability.
+ if (mainOperator instanceof SupportsSoftBackpressure) {
+ SupportsSoftBackpressure head = (SupportsSoftBackpressure)
mainOperator;
Review Comment:
If we go with the replacement, this field would need to never be null:
assign `AvailabilityProvider.AVAILABLE` when the chain has no network outputs /
is finished, and move the assignment out of `if (operatorFactory != null)`.
That matches `NonRecordWriter` (isAvailable always true) and removes the null
checks in StreamTask. One prerequisite worth pinning with tests: the
traversal-collected writer set == the recordWriter delegate's set (side
outputs, multiple-input); a missed writer would silently degrade to hard
blocking.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -328,6 +328,8 @@ public abstract class StreamTask<OUT, OP extends
StreamOperator<OUT>>
@Nullable private final AvailabilityProvider
changelogWriterAvailabilityProvider;
+ @Nullable private AvailabilityProvider operatorChainAvailabilityProvider =
null;
Review Comment:
With the replacement semantics this field is "the task's single output-side
availability" (record writers are just the default leaves), so
`outputAvailabilityProvider` might read better than
`operatorChainAvailabilityProvider`.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SupportsSoftBackpressure.java:
##########
@@ -0,0 +1,38 @@
+/*
+ * 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.flink.streaming.api.operators;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.io.AvailabilityProvider;
+
+/**
+ * Interface for operators that absorb downstream availability internally,
handling back-pressure by
+ * deferring emits and yielding the mailbox rather than blocking.
+ *
+ * <p>{@code OperatorChain} injects the composite downstream {@link
AvailabilityProvider} via {@link
+ * #setDownstreamAvailabilityProvider} and lets the operator represent the
chain's availability.
+ */
+@Internal
+public interface SupportsSoftBackpressure extends AvailabilityProvider {
+
+ /**
+ * Called once by {@code OperatorChain} during construction to inject
downstream availability.
+ */
+ void setDownstreamAvailabilityProvider(AvailabilityProvider provider);
Review Comment:
`SupportsSoftBackpressure` describes the effect rather than the contract —
which IIUC is: the
operator's availability *replaces* its downstream subtree's, and in
exchange it must defer emissions while downstream is unavailable. WDYT about
e.g. `SupportsOutputAvailability` with that spelled out in the javadoc?
--
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]
