Method SetupConnection

be/src/rpc/TAcceptQueueServer.cpp:230–330

New.

// New.
void TAcceptQueueServer::SetupConnection(TAcceptQueueEntry* entry) {
  DCHECK(entry != nullptr);
  if (metrics_enabled_) queue_size_metric_->Increment(-1);
  shared_ptr<TTransport> io_transport;
  shared_ptr<TTransport> client = entry->client_;
  int64_t max_message_size = is_external_facing_ ? ThriftExternalRpcMaxMessageSize() :
      ThriftInternalRpcMaxMessageSize();
  SetMaxMessageSize(client.get(), max_message_size);
  const string& socket_info = reinterpret_cast<TSocket*>(client.get())->getSocketInfo();
  VLOG(2) << Substitute("TAcceptQueueServer: $0 started connection setup for client $1",
      name_, socket_info);
  try {
    MonotonicStopWatch timer;
    // Start timing for connection setup.
    timer.Start();

    // Since THRIFT-5237, it is necessary for Impala to have the same TTransport object
    // for both input and output transport. The detailed reasoning on why this TTransport
    // object sharing requirement is as follow:
    // - Thrift decrements the max message size counter as messages arrive.
    // - Thrift resets the max message size counter with a flush.
    // - If the input and output transport are distinct, the decrement is happening on
    //   one object while the reset is happening on a different object, so it eventually
    //   throws an error.
    // Using same transport fixes the counter logic. This also helps with simplifying
    // Impala's custom TSaslTransport since its caching algorithm in
    // TSaslServerTransport::Factory is not required anymore.
    DCHECK(inputTransportFactory_ == outputTransportFactory_);
    io_transport = inputTransportFactory_->getTransport(client);
    DCHECK_EQ(io_transport->getConfiguration()->getMaxMessageSize(),
        client->getConfiguration()->getMaxMessageSize());
    DCHECK_EQ(max_message_size, io_transport->getConfiguration()->getMaxMessageSize());

    shared_ptr<TProtocol> inputProtocol =
        inputProtocolFactory_->getProtocol(io_transport);
    shared_ptr<TProtocol> outputProtocol =
        outputProtocolFactory_->getProtocol(io_transport);
    shared_ptr<TProcessor> processor =
        getProcessor(inputProtocol, outputProtocol, client);

    if (metrics_enabled_) {
      cnxns_setup_time_us_metric_->Update(timer.ElapsedTime() / NANOS_PER_MICRO);
    }
    VLOG(2) << Substitute("TAcceptQueueServer: $0 finished connection setup for "
        "client $1", name_, socket_info);

    TAcceptQueueServer::Task* task = new TAcceptQueueServer::Task(
        *this, processor, inputProtocol, outputProtocol, client);

    // Create a task
    shared_ptr<Runnable> runnable = shared_ptr<Runnable>(task);

    // Create a thread for this task
    shared_ptr<Thread> thread = shared_ptr<Thread>(threadFactory_->newThread(runnable));

    // Insert thread into the set of threads.
    // Start timing the wait duration for service threads.
    timer.Reset();