You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "rkhachatryan (via GitHub)" <gi...@apache.org> on 2026/02/17 11:55:40 UTC
[PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
rkhachatryan opened a new pull request, #27624:
URL: https://github.com/apache/flink/pull/27624
(no comment)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #27624:
URL: https://github.com/apache/flink/pull/27624#discussion_r2817746171
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/AbstractStreamTaskNetworkInput.java:
##########
@@ -318,10 +324,17 @@ public void close() throws IOException {
// release the deserializers . this part should not ever fail
Exception err = null;
for (InputChannelInfo channelInfo : new ArrayList<>(recordDeserializers.keySet())) {
+ final boolean hadError =
+ checkpointedInputGate.getChannel(channelInfo.getInputChannelIdx()).hasError();
Review Comment:
In theory, recycling the buffer can also fail for other reason:
```
if (inputChannel.isReleased()) {
globalPool.recycleUnpooledMemorySegments(Collections.singletonList(segment));
```
But I think global pool failure shouldn't be ignored.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #27624:
URL: https://github.com/apache/flink/pull/27624#discussion_r2826895602
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/AbstractStreamTaskNetworkInput.java:
##########
@@ -316,8 +322,23 @@ public CompletableFuture<?> getAvailableFuture() {
@Override
public void close() throws IOException {
// release the deserializers . this part should not ever fail
Review Comment:
Good point, updated the comment.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #27624:
URL: https://github.com/apache/flink/pull/27624#discussion_r2826888694
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/AbstractStreamTaskNetworkInput.java:
##########
@@ -316,8 +322,23 @@ public CompletableFuture<?> getAvailableFuture() {
@Override
public void close() throws IOException {
// release the deserializers . this part should not ever fail
+ Exception err = null;
for (InputChannelInfo channelInfo : new ArrayList<>(recordDeserializers.keySet())) {
- releaseDeserializer(channelInfo);
+ final boolean hadError =
+ checkpointedInputGate.getChannel(channelInfo.getInputChannelIdx()).hasError();
+ try {
+ releaseDeserializer(channelInfo);
+ } catch (Exception e) {
+ if (hadError) {
+ LOG.warn(
+ "Ignoring deserializer release failure - the channel has encountered an error before");
+ } else {
+ err = e;
Review Comment:
Good point, I'll use `firstOrSuppressed`.
Covering this with unti tests I think is overkill though, this is not a critical information IMO
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #27624:
URL: https://github.com/apache/flink/pull/27624#discussion_r2817104525
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannel.java:
##########
@@ -346,6 +346,11 @@ public long unsynchronizedGetSizeOfQueuedBuffers() {
*/
public void notifyRequiredSegmentId(int subpartitionId, int segmentId) throws IOException {}
+ /** Whether this input channel has encountered error. */
+ public boolean hasError() {
+ return cause.get() != null;
Review Comment:
I agree, I just wanted to agree on the approach first.
Other approaches are passing `ignoreErrors` to `notifyBufferAvailable` or ignoring errors coming from it.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #27624:
URL: https://github.com/apache/flink/pull/27624#discussion_r2817104525
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannel.java:
##########
@@ -346,6 +346,11 @@ public long unsynchronizedGetSizeOfQueuedBuffers() {
*/
public void notifyRequiredSegmentId(int subpartitionId, int segmentId) throws IOException {}
+ /** Whether this input channel has encountered error. */
+ public boolean hasError() {
+ return cause.get() != null;
Review Comment:
I agree, I just wanted to agree on the approach first.
Other approaches are passing `ignoreErrors` to `notifyBufferAvailable` or ignoring errors from it.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #27624:
URL: https://github.com/apache/flink/pull/27624#discussion_r2817151913
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferManager.java:
##########
@@ -216,6 +221,10 @@ public void recycle(MemorySegment segment) {
if (inputChannel.isReleased()) {
globalPool.recycleUnpooledMemorySegments(Collections.singletonList(segment));
return;
+ } else if (inputChannel.hasError()) {
+ LOG.warn("Input channel has errors - recycling the buffer");
Review Comment:
IIRC the script greps errors, not warnings, so it should be fine.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "Izeren (via GitHub)" <gi...@apache.org>.
Izeren commented on code in PR #27624:
URL: https://github.com/apache/flink/pull/27624#discussion_r2816731122
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferManager.java:
##########
@@ -216,6 +221,10 @@ public void recycle(MemorySegment segment) {
if (inputChannel.isReleased()) {
globalPool.recycleUnpooledMemorySegments(Collections.singletonList(segment));
return;
+ } else if (inputChannel.hasError()) {
+ LOG.warn("Input channel has errors - recycling the buffer");
+ globalPool.recycleUnpooledMemorySegments(Collections.singletonList(segment));
Review Comment:
For my understanding, does every segment have only 1 input channel?
Can it happen that we would call `recycleUnpooledMemorySegments` on the same segment twice? If yes, is this call idempotent?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannel.java:
##########
@@ -346,6 +346,11 @@ public long unsynchronizedGetSizeOfQueuedBuffers() {
*/
public void notifyRequiredSegmentId(int subpartitionId, int segmentId) throws IOException {}
+ /** Whether this input channel has encountered error. */
+ public boolean hasError() {
+ return cause.get() != null;
Review Comment:
If it is not too complicated would be good to write a dedicated unit test for new public method
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "Izeren (via GitHub)" <gi...@apache.org>.
Izeren commented on code in PR #27624:
URL: https://github.com/apache/flink/pull/27624#discussion_r2816722535
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferManager.java:
##########
@@ -216,6 +221,10 @@ public void recycle(MemorySegment segment) {
if (inputChannel.isReleased()) {
globalPool.recycleUnpooledMemorySegments(Collections.singletonList(segment));
return;
+ } else if (inputChannel.hasError()) {
+ LOG.warn("Input channel has errors - recycling the buffer");
Review Comment:
Can this warn result in e2e test failure as "unexpected error" or do we only filter for ERROR level there?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #27624:
URL: https://github.com/apache/flink/pull/27624#discussion_r2826891484
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/AbstractStreamTaskNetworkInput.java:
##########
@@ -316,8 +322,23 @@ public CompletableFuture<?> getAvailableFuture() {
@Override
public void close() throws IOException {
// release the deserializers . this part should not ever fail
+ Exception err = null;
for (InputChannelInfo channelInfo : new ArrayList<>(recordDeserializers.keySet())) {
- releaseDeserializer(channelInfo);
+ final boolean hadError =
+ checkpointedInputGate.getChannel(channelInfo.getInputChannelIdx()).hasError();
+ try {
+ releaseDeserializer(channelInfo);
+ } catch (Exception e) {
+ if (hadError) {
+ LOG.warn(
+ "Ignoring deserializer release failure - the channel has encountered an error before");
Review Comment:
Yeah, there' s actually `channelInfo` available - I'll add it to the log message.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on PR #27624:
URL: https://github.com/apache/flink/pull/27624#issuecomment-3935602579
Thanks for the reviews @1996fanrui @Izeren!
The previous build failed due to FLINK-38613.
I'm going to proceed with merging
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on PR #27624:
URL: https://github.com/apache/flink/pull/27624#issuecomment-3932800452
> According the inline comment, I wonder if we should fix this at a different layer.
Yeah, I'm not sure which level is better to fix this issue.
In the original version, I fixed it at a lower level as you suggest. But there, it affects other notification paths besides of Task cleanup - which probably should **not** tolerate this error.
In the current version, we're certain that it's Task cleanup, so the error can be ignored (if the channel errored already). However, it can be some error not related to channel.
Maybe the middle ground is to catch it StreamTaskNetworkInput but check explicitly that it's a `RemoteTransportException`.
Something like
```
} catch (Exception e) {
if (hadError &&
ExceptionUtils.findThrowable(e, RemoteTransportException.class).isPresent()) {
```
WDYT?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #27624:
URL: https://github.com/apache/flink/pull/27624#issuecomment-3914344758
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ebb28cf1c1d330ee8e0b9aa7f11dc58ef06a37c6",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ebb28cf1c1d330ee8e0b9aa7f11dc58ef06a37c6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ebb28cf1c1d330ee8e0b9aa7f11dc58ef06a37c6 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan merged PR #27624:
URL: https://github.com/apache/flink/pull/27624
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on PR #27624:
URL: https://github.com/apache/flink/pull/27624#issuecomment-3935022883
> In the original version, I fixed it at a lower level as you suggest. But there, it affects other notification paths besides of Task cleanup - which probably should **not** tolerate this error.
Thanks for the valuable input! I did not test it, so not sure if the notification paths is suitable, or they can be changed as well.
> In the current version, we're certain that it's Task cleanup, so the error can be ignored (if the channel errored already). However, it can be some error not related to channel.
>
> Maybe the middle ground is to catch it StreamTaskNetworkInput but check explicitly that it's a `RemoteTransportException`. Something like
>
> ```
> } catch (Exception e) {
> if (hadError &&
> ExceptionUtils.findThrowable(e, RemoteTransportException.class).isPresent()) {
> ```
>
> WDYT?
The new proposal makes sense to me, it only suppresses the `RemoteTransportException`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "Izeren (via GitHub)" <gi...@apache.org>.
Izeren commented on code in PR #27624:
URL: https://github.com/apache/flink/pull/27624#discussion_r2826737469
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/AbstractStreamTaskNetworkInput.java:
##########
@@ -316,8 +322,23 @@ public CompletableFuture<?> getAvailableFuture() {
@Override
public void close() throws IOException {
// release the deserializers . this part should not ever fail
Review Comment:
🤔, this part shouldn't ever fail and we put try-catch right after. Should we update this comment explaining how can it fail?
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/AbstractStreamTaskNetworkInput.java:
##########
@@ -316,8 +322,23 @@ public CompletableFuture<?> getAvailableFuture() {
@Override
public void close() throws IOException {
// release the deserializers . this part should not ever fail
+ Exception err = null;
for (InputChannelInfo channelInfo : new ArrayList<>(recordDeserializers.keySet())) {
- releaseDeserializer(channelInfo);
+ final boolean hadError =
+ checkpointedInputGate.getChannel(channelInfo.getInputChannelIdx()).hasError();
+ try {
+ releaseDeserializer(channelInfo);
+ } catch (Exception e) {
+ if (hadError) {
+ LOG.warn(
+ "Ignoring deserializer release failure - the channel has encountered an error before");
+ } else {
+ err = e;
Review Comment:
!nit, If multiple channels throw, we will only capture the last one. Maybe it would make sense to chain them like: `err = ExceptionUtils.firstOrSuppressed(e, err);` or similar, WDYT
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/AbstractStreamTaskNetworkInput.java:
##########
@@ -316,8 +322,23 @@ public CompletableFuture<?> getAvailableFuture() {
@Override
public void close() throws IOException {
// release the deserializers . this part should not ever fail
+ Exception err = null;
for (InputChannelInfo channelInfo : new ArrayList<>(recordDeserializers.keySet())) {
- releaseDeserializer(channelInfo);
+ final boolean hadError =
+ checkpointedInputGate.getChannel(channelInfo.getInputChannelIdx()).hasError();
+ try {
+ releaseDeserializer(channelInfo);
+ } catch (Exception e) {
+ if (hadError) {
+ LOG.warn(
+ "Ignoring deserializer release failure - the channel has encountered an error before");
Review Comment:
Is there any channel info/task info we would like to log, so that we know where this error was coming from?
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/AbstractStreamTaskNetworkInput.java:
##########
@@ -316,8 +322,23 @@ public CompletableFuture<?> getAvailableFuture() {
@Override
public void close() throws IOException {
// release the deserializers . this part should not ever fail
+ Exception err = null;
for (InputChannelInfo channelInfo : new ArrayList<>(recordDeserializers.keySet())) {
- releaseDeserializer(channelInfo);
+ final boolean hadError =
+ checkpointedInputGate.getChannel(channelInfo.getInputChannelIdx()).hasError();
+ try {
+ releaseDeserializer(channelInfo);
+ } catch (Exception e) {
+ if (hadError) {
+ LOG.warn(
+ "Ignoring deserializer release failure - the channel has encountered an error before");
+ } else {
+ err = e;
Review Comment:
If we do so, may make sense to add unit test covering multiple failures
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #27624:
URL: https://github.com/apache/flink/pull/27624#discussion_r2817748641
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannel.java:
##########
@@ -346,6 +346,11 @@ public long unsynchronizedGetSizeOfQueuedBuffers() {
*/
public void notifyRequiredSegmentId(int subpartitionId, int segmentId) throws IOException {}
+ /** Whether this input channel has encountered error. */
+ public boolean hasError() {
+ return cause.get() != null;
Review Comment:
Added a test
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on PR #27624:
URL: https://github.com/apache/flink/pull/27624#issuecomment-3916907820
CI failures seem unrelated
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #27624:
URL: https://github.com/apache/flink/pull/27624#discussion_r2817140599
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferManager.java:
##########
@@ -216,6 +221,10 @@ public void recycle(MemorySegment segment) {
if (inputChannel.isReleased()) {
globalPool.recycleUnpooledMemorySegments(Collections.singletonList(segment));
return;
+ } else if (inputChannel.hasError()) {
+ LOG.warn("Input channel has errors - recycling the buffer");
+ globalPool.recycleUnpooledMemorySegments(Collections.singletonList(segment));
Review Comment:
Yes, at any given time each memory segment has only one owner, and recycling it twice would cause an 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "Izeren (via GitHub)" <gi...@apache.org>.
Izeren commented on code in PR #27624:
URL: https://github.com/apache/flink/pull/27624#discussion_r2827832203
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/AbstractStreamTaskNetworkInput.java:
##########
@@ -315,9 +321,28 @@ public CompletableFuture<?> getAvailableFuture() {
@Override
public void close() throws IOException {
- // release the deserializers . this part should not ever fail
+ // WARN: throwing an exception from this method might fail Task close procedure and
+ // termination th TM
Review Comment:
!nit, "of the"?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39103][runtime] Tolerate errors when recycling buffer from input channel notification if it had errors already [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #27624:
URL: https://github.com/apache/flink/pull/27624#discussion_r2828119194
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/AbstractStreamTaskNetworkInput.java:
##########
@@ -318,10 +324,17 @@ public void close() throws IOException {
// release the deserializers . this part should not ever fail
Exception err = null;
for (InputChannelInfo channelInfo : new ArrayList<>(recordDeserializers.keySet())) {
+ final boolean hadError =
+ checkpointedInputGate.getChannel(channelInfo.getInputChannelIdx()).hasError();
Review Comment:
I share the same concern you mentioned.
The current `catch (Exception e) + if (hadError)` would swallow *all* exceptions from `releaseDeserializer`, not just the channel error. If e.g. `globalPool.recycleUnpooledMemorySegments` fails while `hadError=true`, that failure is silently lost.
And it is dangerous if introducing new exception in the future.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org