ARROW-3292: [C++] Test Flight RPC in Travis CI

.travis.yml

@@ -62,6 +62,7 @@ matrix:
     env:
     - ARROW_TRAVIS_VALGRIND=1
     - ARROW_TRAVIS_USE_TOOLCHAIN=1
+    - ARROW_TRAVIS_FLIGHT=1
     - ARROW_TRAVIS_PLASMA=1
     - ARROW_TRAVIS_ORC=1
     - ARROW_TRAVIS_PARQUET=1
@@ -86,10 +87,11 @@ matrix:
     os: linux
     jdk: openjdk8
     env:
+    - ARROW_TRAVIS_COVERAGE=1
     - ARROW_TRAVIS_USE_TOOLCHAIN=1
+    - ARROW_TRAVIS_FLIGHT=1
     - ARROW_TRAVIS_PLASMA=1
     - ARROW_TRAVIS_ORC=1
-    - ARROW_TRAVIS_COVERAGE=1
     - ARROW_TRAVIS_PARQUET=1
     - ARROW_TRAVIS_GANDIVA=1
     - ARROW_TRAVIS_GANDIVA_JAVA=1
@@ -181,6 +183,7 @@ matrix:
     env:
     - ARROW_TRAVIS_USE_TOOLCHAIN=1
     - ARROW_TRAVIS_PLASMA=1
+    - ARROW_TRAVIS_FLIGHT=1
     - ARROW_TRAVIS_ORC=1
     - ARROW_TRAVIS_PARQUET=1
     - ARROW_TRAVIS_GANDIVA=1

ci/conda_env_cpp.yml

@@ -20,12 +20,14 @@
 boost-cpp=1.68.0
 brotli
 bzip2
+c-ares
 cmake
 double-conversion
 flatbuffers
 gflags
 glog
 gmock
+grpc-cpp
 gtest
 libprotobuf
 lz4-c

ci/travis_before_script_cpp.sh

@@ -86,6 +86,10 @@ if [ "$ARROW_TRAVIS_USE_TOOLCHAIN" == "1" ]; then
   fi
 fi
 
+if [ "$ARROW_TRAVIS_FLIGHT" == "1" ]; then
+  CMAKE_COMMON_FLAGS="$CMAKE_COMMON_FLAGS -DARROW_FLIGHT=ON"
+fi
+
 if [ "$ARROW_TRAVIS_PLASMA" == "1" ]; then
   CMAKE_COMMON_FLAGS="$CMAKE_COMMON_FLAGS -DARROW_PLASMA=ON"
 fi

cpp/cmake_modules/Findc-ares.cmake

@@ -42,8 +42,9 @@ if (MSVC)
 else ()
   set(CARES_LIB_NAME
     ${CMAKE_SHARED_LIBRARY_PREFIX}cares${CMAKE_SHARED_LIBRARY_SUFFIX})
-  set(CARES_STATIC_LIB_NAME
-    ${CMAKE_STATIC_LIBRARY_PREFIX}cares${CMAKE_STATIC_LIBRARY_SUFFIX})
+  set(CARES_STATIC_LIB_NAMES
+    ${CMAKE_STATIC_LIBRARY_PREFIX}cares${CMAKE_STATIC_LIBRARY_SUFFIX}
+    ${CMAKE_STATIC_LIBRARY_PREFIX}cares_static${CMAKE_STATIC_LIBRARY_SUFFIX})
 endif ()
 
 # Try the parameterized roots, if they exist
@@ -56,7 +57,7 @@ if (_cares_roots)
     PATHS ${_cares_roots} NO_DEFAULT_PATH
     PATH_SUFFIXES "lib")
   find_library(CARES_STATIC_LIB
-    NAMES ${CARES_STATIC_LIB_NAME}
+    NAMES ${CARES_STATIC_LIB_NAMES}
     PATHS ${_cares_roots} NO_DEFAULT_PATH
     PATH_SUFFIXES "lib")
 else ()

cpp/src/arrow/flight/CMakeLists.txt

@@ -28,12 +28,11 @@ set(ARROW_FLIGHT_STATIC_LINK_LIBS
     grpc_address_sorting_static
     cares_static)
 
-set(ARROW_FLIGHT_TEST_STATIC_LINK_LIBS
-    arrow_static
-    arrow_flight_static
-    arrow_testing_static
-    ${ARROW_FLIGHT_STATIC_LINK_LIBS}
-    ${PROTOBUF_LIBRARY})
+set(ARROW_FLIGHT_TEST_LINK_LIBS
+    arrow_flight_shared
+    arrow_flight_testing_shared
+    ${ARROW_TEST_SHARED_LINK_LIBS}
+    ${ARROW_FLIGHT_STATIC_LINK_LIBS})
 
 # TODO(wesm): Protobuf shared vs static linking
 
@@ -78,25 +77,43 @@ add_arrow_lib(arrow_flight
               arrow_static
               ${ARROW_FLIGHT_STATIC_LINK_LIBS})
 
+# Define arrow_flight_testing library
+if(ARROW_BUILD_TESTS OR ARROW_BUILD_BENCHMARKS)
+  add_arrow_lib(arrow_flight_testing
+                SOURCES
+                test-util.cc
+                DEPENDENCIES
+                ${GTEST_LIBRARY}
+                SHARED_LINK_LIBS
+                arrow_shared
+                arrow_flight_shared
+                ${BOOST_FILESYSTEM_LIBRARY}
+                ${BOOST_SYSTEM_LIBRARY}
+                ${GTEST_LIBRARY}
+                STATIC_LINK_LIBS
+                arrow_static
+                arrow_flight_static)
+endif()
+
 add_arrow_test(flight-test
                EXTRA_LINK_LIBS
-               ${ARROW_FLIGHT_TEST_STATIC_LINK_LIBS}
+               ${ARROW_FLIGHT_TEST_LINK_LIBS}
                LABELS
                "arrow_flight")
 
 # Build test server for unit tests or benchmarks
 if(ARROW_BUILD_TESTS OR ARROW_BUILD_BENCHMARKS)
   add_executable(flight-test-server test-server.cc)
-  target_link_libraries(flight-test-server ${ARROW_FLIGHT_TEST_STATIC_LINK_LIBS}
-                        gflags_static ${GTEST_LIBRARY})
+  target_link_libraries(flight-test-server ${ARROW_FLIGHT_TEST_LINK_LIBS} gflags_static
+                        ${GTEST_LIBRARY})
 
   add_executable(flight-test-integration-server test-integration-server.cc)
-  target_link_libraries(flight-test-integration-server
-                        ${ARROW_FLIGHT_TEST_STATIC_LINK_LIBS} gflags_static gtest_static)
+  target_link_libraries(flight-test-integration-server ${ARROW_FLIGHT_TEST_LINK_LIBS}
+                        gflags_static gtest_static)
 
   add_executable(flight-test-integration-client test-integration-client.cc)
-  target_link_libraries(flight-test-integration-client
-                        ${ARROW_FLIGHT_TEST_STATIC_LINK_LIBS} gflags_static gtest_static)
+  target_link_libraries(flight-test-integration-client ${ARROW_FLIGHT_TEST_LINK_LIBS}
+                        gflags_static gtest_static)
 
   # This is needed for the unit tests
   if(ARROW_BUILD_TESTS)
@@ -116,17 +133,17 @@ if(ARROW_BUILD_BENCHMARKS)
 
   add_executable(flight-perf-server perf-server.cc perf.pb.cc)
   target_link_libraries(flight-perf-server
-                        arrow_flight_static
-                        arrow_testing_static
-                        ${ARROW_FLIGHT_STATIC_LINK_LIBS}
+                        arrow_flight_shared
+                        arrow_flight_testing_shared
+                        ${ARROW_FLIGHT_TEST_LINK_LIBS}
                         gflags_static
                         ${GTEST_LIBRARY})
 
   add_executable(flight-benchmark flight-benchmark.cc perf.pb.cc)
   target_link_libraries(flight-benchmark
                         arrow_flight_static
                         arrow_testing_static
-                        ${ARROW_FLIGHT_STATIC_LINK_LIBS}
+                        ${ARROW_FLIGHT_TEST_LINK_LIBS}
                         gflags_static
                         ${GTEST_LIBRARY})
 

cpp/src/arrow/flight/client.cc

@@ -196,8 +196,11 @@ class FlightClient::FlightClientImpl {
     ss << host << ":" << port;
     std::string uri = ss.str();
 
+    grpc::ChannelArguments args;
+    // Try to reconnect quickly at first, in case the server is still starting up
+    args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, 100);
     stub_ = pb::FlightService::NewStub(
-        grpc::CreateChannel(ss.str(), grpc::InsecureChannelCredentials()));
+        grpc::CreateCustomChannel(ss.str(), grpc::InsecureChannelCredentials(), args));
     return Status::OK();
   }
 

cpp/src/arrow/flight/client.h

@@ -116,7 +116,7 @@ class ARROW_EXPORT FlightClient {
 /// \brief An interface to upload record batches to a Flight server
 class ARROW_EXPORT FlightPutWriter : public ipc::RecordBatchWriter {
  public:
-  ~FlightPutWriter();
+  ~FlightPutWriter() override;
 
   Status WriteRecordBatch(const RecordBatch& batch, bool allow_64bit = false) override;
   Status Close() override;

cpp/src/arrow/flight/flight-test.cc

@@ -28,10 +28,10 @@
 #include <iostream>
 #include <memory>
 #include <string>
+#include <thread>
 #include <vector>
 
 #include <gtest/gtest.h>
-#include <boost/process.hpp>
 
 #include "arrow/ipc/test-common.h"
 #include "arrow/status.h"
@@ -62,6 +62,7 @@ TEST(TestFlight, StartStopTestServer) {
   ASSERT_TRUE(server.IsRunning());
   int exit_code = server.Stop();
   ASSERT_EQ(0, exit_code);
+  ASSERT_FALSE(server.IsRunning());
 }
 
 // ----------------------------------------------------------------------

cpp/src/arrow/flight/perf-server.cc

@@ -29,9 +29,11 @@
 #include "arrow/io/test-common.h"
 #include "arrow/ipc/writer.h"
 #include "arrow/record_batch.h"
+#include "arrow/testing/util.h"
 
+#include "arrow/flight/api.h"
+#include "arrow/flight/internal.h"
 #include "arrow/flight/perf.pb.h"
-#include "arrow/flight/server.h"
 #include "arrow/flight/test-util.h"
 
 DEFINE_int32(port, 31337, "Server port to listen on");

cpp/src/arrow/flight/serialization-internal.h

@@ -25,6 +25,7 @@
 
 #include "google/protobuf/io/coded_stream.h"
 #include "google/protobuf/io/zero_copy_stream.h"
+#include "google/protobuf/io/zero_copy_stream_impl_lite.h"
 #include "google/protobuf/wire_format_lite.h"
 #include "grpc/byte_buffer_reader.h"
 #include "grpcpp/grpcpp.h"
@@ -66,33 +67,6 @@ namespace internal {
 using google::protobuf::io::CodedInputStream;
 using google::protobuf::io::CodedOutputStream;
 
-// More efficient writing of FlightData to gRPC output buffer
-// Implementation of ZeroCopyOutputStream that writes to a fixed-size buffer
-class FixedSizeProtoWriter : public ::google::protobuf::io::ZeroCopyOutputStream {
- public:
-  explicit FixedSizeProtoWriter(grpc_slice slice)
-      : slice_(slice),
-        bytes_written_(0),
-        total_size_(static_cast<int>(GRPC_SLICE_LENGTH(slice))) {}
-
-  bool Next(void** data, int* size) override {
-    // Consume the whole slice
-    *data = GRPC_SLICE_START_PTR(slice_) + bytes_written_;
-    *size = total_size_ - bytes_written_;
-    bytes_written_ = total_size_;
-    return true;
-  }
-
-  void BackUp(int count) override { bytes_written_ -= count; }
-
-  int64_t ByteCount() const override { return bytes_written_; }
-
- private:
-  grpc_slice slice_;
-  int bytes_written_;
-  int total_size_;
-};
-
 bool ReadBytesZeroCopy(const std::shared_ptr<arrow::Buffer>& source_data,
                        CodedInputStream* input, std::shared_ptr<arrow::Buffer>* out);
 
@@ -159,11 +133,11 @@ class GrpcBuffer : public arrow::MutableBuffer {
 namespace grpc {
 
 using arrow::flight::FlightData;
-using arrow::flight::internal::FixedSizeProtoWriter;
 using arrow::flight::internal::GrpcBuffer;
 using arrow::flight::internal::ReadBytesZeroCopy;
 
 using google::protobuf::internal::WireFormatLite;
+using google::protobuf::io::ArrayOutputStream;
 using google::protobuf::io::CodedInputStream;
 using google::protobuf::io::CodedOutputStream;
 
@@ -298,7 +272,8 @@ class SerializationTraits<IpcPayload> {
     // XXX(wesm): for debugging
     // std::cout << "Writing record batch with total size " << total_size << std::endl;
 
-    FixedSizeProtoWriter writer(*reinterpret_cast<grpc_slice*>(&slice));
+    ArrayOutputStream writer(const_cast<uint8_t*>(slice.begin()),
+                             static_cast<int>(slice.size()));
     CodedOutputStream pb_stream(&writer);
 
     // Write header

cpp/src/arrow/flight/test-integration-client.cc

@@ -29,10 +29,12 @@
 
 #include "arrow/io/test-common.h"
 #include "arrow/ipc/json.h"
+#include "arrow/ipc/writer.h"
 #include "arrow/record_batch.h"
 #include "arrow/table.h"
+#include "arrow/util/logging.h"
 
-#include "arrow/flight/server.h"
+#include "arrow/flight/api.h"
 #include "arrow/flight/test-util.h"
 
 DEFINE_string(host, "localhost", "Server port to connect to");

cpp/src/arrow/flight/test-integration-server.cc

@@ -28,7 +28,9 @@
 #include "arrow/ipc/json.h"
 #include "arrow/record_batch.h"
 #include "arrow/table.h"
+#include "arrow/util/logging.h"
 
+#include "arrow/flight/internal.h"
 #include "arrow/flight/server.h"
 #include "arrow/flight/test-util.h"