code: public ClientConnection createFlow(final Flow flow, final ReplyProcessor rp) {
          Exceptions.checkNotClosed(closed.get(), this);
          Preconditions.checkState(!disableFlow.get(), "Ensure flows are enabled.");
          log.info("Creating Flow {} for endpoint {}. The current Channel is {}.", flow.getFlowId(), connectionName, channel.get());
          if (flowIdReplyProcessorMap.put(flow.getFlowId(), rp) != null) {
              throw new IllegalArgumentException("Multiple flows cannot be created with the same Flow id " + flow.getFlowId());
          }
          return new ClientConnectionImpl(connectionName, flow.getFlowId(), batchSizeTracker, this);
      }
docs: Create a flow on existing connection.
  @param flow Flow.
  @param rp ReplyProcessor for the specified flow.
  @return Client Connection object.