[NO ISSUE][NET] Ensure Channel Was Not Recycled Before Adding Credit

- user model changes: no
- storage format changes: no
- interface changes: no

Details:
- Before attempting to add credit to an existing channel,
  ensure the channel slot was not recycled and reused by a
  new channel. Without this, some additional credit might
  erroneously be added to the new channel.

Change-Id: I0a5ec70ad4c2568477fac9eb98c030fc3d8b1ce9
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/4623
Integration-Tests: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Reviewed-by: Michael Blow <mblow@apache.org>
Contrib: Michael Blow <mblow@apache.org>
diff --git a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelControlBlock.java b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelControlBlock.java
index 75d442f..f7b4fbd 100644
--- a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelControlBlock.java
+++ b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelControlBlock.java
@@ -144,7 +144,7 @@
 
     @Override
     public void addPendingCredits(int credit) {
-        cSet.addPendingCredits(channelId, credit);
+        cSet.addPendingCredits(this, credit);
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelSet.java b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelSet.java
index f5cdf2c..7784fc4 100644
--- a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelSet.java
+++ b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelSet.java
@@ -140,13 +140,15 @@
         }
     }
 
-    void addPendingCredits(int channelId, int delta) {
+    void addPendingCredits(ChannelControlBlock targetCcb, int delta) {
         if (delta <= 0) {
             return;
         }
         synchronized (mConn) {
+            final int channelId = targetCcb.getChannelId();
             ChannelControlBlock ccb = ccbArray[channelId];
-            if (ccb != null) {
+            // ensure the channel slot id was not recycled and used for a diffierent channel
+            if (ccb == targetCcb) {
                 if (ccb.getRemoteEOS()) {
                     return;
                 }