[hotfix][network] Remove the deduplication for CancelPartitionRequest.

To avoid the memory leak in the record map for deduplication.

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandler.java

@@ -67,13 +67,6 @@ class CreditBasedPartitionRequestClientHandler extends ChannelInboundHandlerAdap
     private final ChannelFutureListener writeListener =
             new WriteAndFlushNextMessageIfPossibleListener();
 
-    /**
-     * Set of cancelled partition requests. A request is cancelled iff an input channel is cleared
-     * while data is still coming in for this channel.
-     */
-    private final ConcurrentMap<InputChannelID, InputChannelID> cancelled =
-            new ConcurrentHashMap<>();
-
     /**
      * The channel handler context is initialized in channel active event by netty thread, the
      * context may also be accessed by task thread or canceler thread to cancel partition request
@@ -108,9 +101,7 @@ public void cancelRequestFor(InputChannelID inputChannelId) {
             return;
         }
 
-        if (cancelled.putIfAbsent(inputChannelId, inputChannelId) == null) {
-            ctx.writeAndFlush(new NettyMessage.CancelPartitionRequest(inputChannelId));
-        }
+        ctx.writeAndFlush(new NettyMessage.CancelPartitionRequest(inputChannelId));
     }
 
     // ------------------------------------------------------------------------