client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java [160:205]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
  @Override
  public void finishReadRecoveredState() {
    // do-nothing.
  }

  @Override
  public InputChannel getChannel(int channelIndex) {
    return new FakedRemoteInputChannel(channelIndex);
  }

  @Override
  public void sendTaskEvent(TaskEvent event) {
    throw new FlinkRuntimeException("Method should not be called.");
  }

  @Override
  public void resumeConsumption(InputChannelInfo channelInfo) {
    throw new FlinkRuntimeException("Method should not be called.");
  }

  @Override
  public void acknowledgeAllRecordsProcessed(InputChannelInfo inputChannelInfo) {}

  @Override
  public CompletableFuture<Void> getStateConsumedFuture() {
    return CompletableFuture.completedFuture(null);
  }

  @Override
  public String toString() {
    return String.format(
        "ReadGate [owning task: %s, gate index: %d, descriptor: %s]",
        inputGateDelegation.getTaskName(),
        inputGateDelegation.getGateIndex(),
        inputGateDelegation.getGateDescriptor().toString());
  }

  /** Accommodation for the incompleteness of Flink pluggable shuffle service. */
  private class FakedRemoteInputChannel extends RemoteInputChannel {
    FakedRemoteInputChannel(int channelIndex) {
      super(
          new SingleInputGate(
              "",
              inputGateDelegation.getGateIndex(),
              new IntermediateDataSetID(),
              ResultPartitionType.BLOCKING,
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java [172:217]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
  @Override
  public void finishReadRecoveredState() {
    // do-nothing.
  }

  @Override
  public InputChannel getChannel(int channelIndex) {
    return new FakedRemoteInputChannel(channelIndex);
  }

  @Override
  public void sendTaskEvent(TaskEvent event) {
    throw new FlinkRuntimeException("Method should not be called.");
  }

  @Override
  public void resumeConsumption(InputChannelInfo channelInfo) {
    throw new FlinkRuntimeException("Method should not be called.");
  }

  @Override
  public void acknowledgeAllRecordsProcessed(InputChannelInfo inputChannelInfo) {}

  @Override
  public CompletableFuture<Void> getStateConsumedFuture() {
    return CompletableFuture.completedFuture(null);
  }

  @Override
  public String toString() {
    return String.format(
        "ReadGate [owning task: %s, gate index: %d, descriptor: %s]",
        inputGateDelegation.getTaskName(),
        inputGateDelegation.getGateIndex(),
        inputGateDelegation.getGateDescriptor().toString());
  }

  /** Accommodation for the incompleteness of Flink pluggable shuffle service. */
  private class FakedRemoteInputChannel extends RemoteInputChannel {
    FakedRemoteInputChannel(int channelIndex) {
      super(
          new SingleInputGate(
              "",
              inputGateDelegation.getGateIndex(),
              new IntermediateDataSetID(),
              ResultPartitionType.BLOCKING,
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -