be/src/testutil/in-process-servers.cc

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#include "testutil/in-process-servers.h"

#include <boost/scoped_ptr.hpp>
#include <stdlib.h>

#include "statestore/statestore.h"
#include "rpc/thrift-util.h"
#include "rpc/thrift-server.h"
#include "util/network-util.h"
#include "util/webserver.h"
#include "util/default-path-handlers.h"
#include "util/metrics.h"
#include "util/openssl-util.h"
#include "runtime/exec-env.h"
#include "scheduling/scheduler.h"
#include "service/frontend.h"
#include "service/impala-server.h"

#include "common/names.h"

DECLARE_string(ssl_server_certificate);
DECLARE_string(ssl_private_key);
DECLARE_int32(be_port);
DECLARE_int32(krpc_port);

using namespace apache::thrift;
using namespace impala;

Status InProcessImpalaServer::StartWithEphemeralPorts(const string& statestore_host,
    int statestore_port, InProcessImpalaServer** server) {
  // These flags are read directly in several places to find the address of the local
  // backend interface.
  FLAGS_be_port = 0;
  // Thrift server ctor allows port to be set to 0. Not supported with KRPC.
  // So KRPC port must be explicitly set here.
  FLAGS_krpc_port = FindUnusedEphemeralPort();

  // Use wildcard addresses of 0 so that the Thrift servers will pick their own port.
  *server = new InProcessImpalaServer(FLAGS_hostname, 0, FLAGS_krpc_port, 0, 0,
      statestore_host, statestore_port);
  // Start the daemon and check if it works, if not delete the current server object and
  // pick a new set of ports
  return (*server)->StartWithClientServers(0, 0, 0);
}

InProcessImpalaServer::InProcessImpalaServer(const string& hostname, int backend_port,
    int krpc_port, int subscriber_port, int webserver_port, const string& statestore_host,
    int statestore_port)
  : backend_port_(backend_port),
    beeswax_port_(0),
    hs2_port_(0),
    hs2_http_port_(0),
    impala_server_(NULL),
    exec_env_(new ExecEnv(backend_port, krpc_port, subscriber_port, webserver_port,
        statestore_host, statestore_port)) {}

void InProcessImpalaServer::SetCatalogIsReady() {
  DCHECK(impala_server_ != NULL) << "Call Start*() first.";
  exec_env_->frontend()->SetCatalogIsReady();
}

Status InProcessImpalaServer::StartWithClientServers(
    int beeswax_port, int hs2_port, int hs2_http_port) {
  RETURN_IF_ERROR(exec_env_->Init());
  beeswax_port_ = beeswax_port;
  hs2_port_ = hs2_port;
  hs2_http_port_ = hs2_http_port;

  impala_server_.reset(new ImpalaServer(exec_env_.get()));
  SetCatalogIsReady();
  RETURN_IF_ERROR(
      impala_server_->Start(backend_port_, beeswax_port, hs2_port, hs2_http_port_));

  // This flag is read directly in several places to find the address of the local
  // backend interface.
  FLAGS_be_port = impala_server_->GetThriftBackendPort();

  // Wait for up to 1s for the backend server to start
  RETURN_IF_ERROR(WaitForServer(FLAGS_hostname, FLAGS_be_port, 10, 100));
  return Status::OK();
}

Status InProcessImpalaServer::Join() {
  impala_server_->Join();
  return Status::OK();
}

int InProcessImpalaServer::GetBeeswaxPort() const {
  return impala_server_->GetBeeswaxPort();
}

int InProcessImpalaServer::GetHS2Port() const {
  return impala_server_->GetHS2Port();
}