Address review feedback

cpp/src/arrow/flight/CMakeLists.txt

@@ -81,10 +81,10 @@ add_custom_target(flight_grpc_gen ALL DEPENDS ${FLIGHT_GENERATED_PROTO_FILES})
 set(ARROW_FLIGHT_SRCS
     client.cc
     internal.cc
-    protocol_internal.cc
-    serialization_internal.cc
     middleware.cc
     middleware_internal.cc
+    protocol_internal.cc
+    serialization_internal.cc
     server.cc
     server_auth.cc
     types.cc)

cpp/src/arrow/flight/client.cc

@@ -117,7 +117,7 @@ class GrpcClientInterceptorAdapter : public grpc::experimental::Interceptor {
             InterceptionHookPoints::PRE_SEND_INITIAL_METADATA)) {
       GrpcAddCallHeaders add_headers(methods->GetSendInitialMetadata());
       for (const auto& middleware : middleware_) {
-        middleware->SendingHeaders(add_headers);
+        middleware->SendingHeaders(&add_headers);
       }
     }
 
@@ -176,6 +176,8 @@ class GrpcClientInterceptorAdapterFactory
       flight_method = FlightMethod::DoAction;
     } else if (method.ends_with("/ListActions")) {
       flight_method = FlightMethod::ListActions;
+    } else {
+      DCHECK(false) << "Unknown Flight method: " << info->method();
     }
 
     const CallInfo flight_info{flight_method};

cpp/src/arrow/flight/client_middleware.h

@@ -27,12 +27,11 @@
 #include "arrow/status.h"
 
 namespace arrow {
-
 namespace flight {
 
 /// \brief Client-side middleware for a call, instantiated per RPC.
 ///
-/// Middleware should be fast and must be infalliable: there is no way
+/// Middleware should be fast and must be infallible: there is no way
 /// to reject the call or report errors from the middleware instance.
 class ARROW_FLIGHT_EXPORT ClientMiddleware {
  public:
@@ -40,7 +39,7 @@ class ARROW_FLIGHT_EXPORT ClientMiddleware {
 
   /// \brief A callback before headers are sent. Extra headers can be
   /// added, but existing ones cannot be read.
-  virtual void SendingHeaders(AddCallHeaders& outgoing_headers) = 0;
+  virtual void SendingHeaders(AddCallHeaders* outgoing_headers) = 0;
 
   /// \brief A callback when headers are received from the server.
   virtual void ReceivedHeaders(const CallHeaders& incoming_headers) = 0;
@@ -71,5 +70,4 @@ class ARROW_FLIGHT_EXPORT ClientMiddlewareFactory {
 };
 
 }  // namespace flight
-
 }  // namespace arrow

cpp/src/arrow/flight/flight_test.cc

@@ -289,24 +289,47 @@ TEST(TestFlight, BuilderHook) {
 // ----------------------------------------------------------------------
 // Client tests
 
+// Helper to initialize a server and matching client with callbacks to
+// populate options.
+template <typename T, typename... Args>
+Status MakeServer(std::unique_ptr<FlightServerBase>* server,
+                  std::unique_ptr<FlightClient>* client,
+                  std::function<Status(FlightServerOptions*)> make_server_options,
+                  std::function<Status(FlightClientOptions*)> make_client_options,
+                  Args&&... server_args) {
+  Location location;
+  RETURN_NOT_OK(Location::ForGrpcTcp("localhost", 0, &location));
+  *server = arrow::internal::make_unique<T>(std::forward<Args>(server_args)...);
+  FlightServerOptions server_options(location);
+  RETURN_NOT_OK(make_server_options(&server_options));
+  RETURN_NOT_OK((*server)->Init(server_options));
+  Location real_location;
+  RETURN_NOT_OK(Location::ForGrpcTcp("localhost", (*server)->port(), &real_location));
+  FlightClientOptions client_options;
+  RETURN_NOT_OK(make_client_options(&client_options));
+  return FlightClient::Connect(real_location, client_options, client);
+}
+
 class TestFlightClient : public ::testing::Test {
  public:
   void SetUp() {
+    server_ = ExampleTestServer();
+
     Location location;
-    std::unique_ptr<FlightServerBase> server = ExampleTestServer();
-
-    ASSERT_OK(Location::ForGrpcTcp("localhost", ::arrow::GetListenPort(), &location));
+    ASSERT_OK(Location::ForGrpcTcp("localhost", 0, &location));
     FlightServerOptions options(location);
-    ASSERT_OK(server->Init(options));
+    ASSERT_OK(server_->Init(options));
 
-    server_.reset(new InProcessTestServer(std::move(server), location));
-    ASSERT_OK(server_->Start());
     ASSERT_OK(ConnectClient());
   }
 
-  void TearDown() { server_->Stop(); }
+  void TearDown() { ASSERT_OK(server_->Shutdown()); }
 
-  Status ConnectClient() { return FlightClient::Connect(server_->location(), &client_); }
+  Status ConnectClient() {
+    Location location;
+    RETURN_NOT_OK(Location::ForGrpcTcp("localhost", server_->port(), &location));
+    return FlightClient::Connect(location, &client_);
+  }
 
   template <typename EndpointCheckFunc>
   void CheckDoGet(const FlightDescriptor& descr, const BatchVector& expected_batches,
@@ -342,9 +365,8 @@ class TestFlightClient : public ::testing::Test {
   }
 
  protected:
-  int port_;
   std::unique_ptr<FlightClient> client_;
-  std::unique_ptr<InProcessTestServer> server_;
+  std::unique_ptr<FlightServerBase> server_;
 };
 
 class AuthTestServer : public FlightServerBase {
@@ -419,104 +441,71 @@ class MetadataTestServer : public FlightServerBase {
   }
 };
 
-template <typename T>
-class InsecureTestServer : public ::testing::Test {
+class TestMetadata : public ::testing::Test {
  public:
   void SetUp() {
-    Location location;
-    ASSERT_OK(Location::ForGrpcTcp("localhost", 30000, &location));
-
-    std::unique_ptr<FlightServerBase> server(new T);
-    FlightServerOptions options(location);
-    ASSERT_OK(server->Init(options));
-
-    server_.reset(new InProcessTestServer(std::move(server), location));
-    ASSERT_OK(server_->Start());
-    ASSERT_OK(ConnectClient());
+    ASSERT_OK(MakeServer<MetadataTestServer>(
+        &server_, &client_, [](FlightServerOptions* options) { return Status::OK(); },
+        [](FlightClientOptions* options) { return Status::OK(); }));
   }
 
-  void TearDown() { server_->Stop(); }
-
-  Status ConnectClient() { return FlightClient::Connect(server_->location(), &client_); }
+  void TearDown() { ASSERT_OK(server_->Shutdown()); }
 
  protected:
-  int port_;
   std::unique_ptr<FlightClient> client_;
-  std::unique_ptr<InProcessTestServer> server_;
+  std::unique_ptr<FlightServerBase> server_;
 };
 
-using TestMetadata = InsecureTestServer<MetadataTestServer>;
-
 class TestAuthHandler : public ::testing::Test {
  public:
   void SetUp() {
-    Location location;
-    std::unique_ptr<FlightServerBase> server(new AuthTestServer);
-
-    ASSERT_OK(Location::ForGrpcTcp("localhost", ::arrow::GetListenPort(), &location));
-    FlightServerOptions options(location);
-    options.auth_handler =
-        std::unique_ptr<ServerAuthHandler>(new TestServerAuthHandler("user", "p4ssw0rd"));
-    ASSERT_OK(server->Init(options));
-
-    server_.reset(new InProcessTestServer(std::move(server), location));
-    ASSERT_OK(server_->Start());
-    ASSERT_OK(ConnectClient());
+    ASSERT_OK(MakeServer<AuthTestServer>(
+        &server_, &client_,
+        [](FlightServerOptions* options) {
+          options->auth_handler = std::unique_ptr<ServerAuthHandler>(
+              new TestServerAuthHandler("user", "p4ssw0rd"));
+          return Status::OK();
+        },
+        [](FlightClientOptions* options) { return Status::OK(); }));
   }
 
-  void TearDown() { server_->Stop(); }
-
-  Status ConnectClient() { return FlightClient::Connect(server_->location(), &client_); }
+  void TearDown() { ASSERT_OK(server_->Shutdown()); }
 
  protected:
   std::unique_ptr<FlightClient> client_;
-  std::unique_ptr<InProcessTestServer> server_;
+  std::unique_ptr<FlightServerBase> server_;
 };
 
 class TestBasicAuthHandler : public ::testing::Test {
  public:
   void SetUp() {
-    Location location;
-    std::unique_ptr<FlightServerBase> server(new AuthTestServer);
-
-    ASSERT_OK(Location::ForGrpcTcp("localhost", GetListenPort(), &location));
-    FlightServerOptions options(location);
-    options.auth_handler = std::unique_ptr<ServerAuthHandler>(
-        new TestServerBasicAuthHandler("user", "p4ssw0rd"));
-    ASSERT_OK(server->Init(options));
-
-    server_.reset(new InProcessTestServer(std::move(server), location));
-    ASSERT_OK(server_->Start());
-    ASSERT_OK(ConnectClient());
+    ASSERT_OK(MakeServer<AuthTestServer>(
+        &server_, &client_,
+        [](FlightServerOptions* options) {
+          options->auth_handler = std::unique_ptr<ServerAuthHandler>(
+              new TestServerBasicAuthHandler("user", "p4ssw0rd"));
+          return Status::OK();
+        },
+        [](FlightClientOptions* options) { return Status::OK(); }));
   }
 
-  void TearDown() { server_->Stop(); }
-
-  Status ConnectClient() { return FlightClient::Connect(server_->location(), &client_); }
+  void TearDown() { ASSERT_OK(server_->Shutdown()); }
 
  protected:
   std::unique_ptr<FlightClient> client_;
-  std::unique_ptr<InProcessTestServer> server_;
+  std::unique_ptr<FlightServerBase> server_;
 };
 
 class TestDoPut : public ::testing::Test {
  public:
   void SetUp() {
-    Location location;
-    ASSERT_OK(Location::ForGrpcTcp("localhost", ::arrow::GetListenPort(), &location));
-
-    do_put_server_ = new DoPutTestServer();
-    server_.reset(new InProcessTestServer(
-        std::unique_ptr<FlightServerBase>(do_put_server_), location));
-    FlightServerOptions options(location);
-    ASSERT_OK(do_put_server_->Init(options));
-    ASSERT_OK(server_->Start());
-    ASSERT_OK(ConnectClient());
+    ASSERT_OK(MakeServer<DoPutTestServer>(
+        &server_, &client_, [](FlightServerOptions* options) { return Status::OK(); },
+        [](FlightClientOptions* options) { return Status::OK(); }));
+    do_put_server_ = (DoPutTestServer*)server_.get();
   }
 
-  void TearDown() { server_->Stop(); }
-
-  Status ConnectClient() { return FlightClient::Connect(server_->location(), &client_); }
+  void TearDown() { ASSERT_OK(server_->Shutdown()); }
 
   void CheckBatches(FlightDescriptor expected_descriptor,
                     const BatchVector& expected_batches) {
@@ -543,46 +532,43 @@ class TestDoPut : public ::testing::Test {
 
  protected:
   std::unique_ptr<FlightClient> client_;
-  std::unique_ptr<InProcessTestServer> server_;
+  std::unique_ptr<FlightServerBase> server_;
   DoPutTestServer* do_put_server_;
 };
 
 class TestTls : public ::testing::Test {
  public:
   void SetUp() {
+    server_.reset(new TlsTestServer);
+
     Location location;
-    std::unique_ptr<FlightServerBase> server(new TlsTestServer);
-
-    ASSERT_OK(Location::ForGrpcTls("localhost", ::arrow::GetListenPort(), &location));
+    ASSERT_OK(Location::ForGrpcTls("localhost", 0, &location));
     FlightServerOptions options(location);
-    ASSERT_RAISES(UnknownError, server->Init(options));
+    ASSERT_RAISES(UnknownError, server_->Init(options));
     ASSERT_OK(ExampleTlsCertificates(&options.tls_certificates));
-    ASSERT_OK(server->Init(options));
+    ASSERT_OK(server_->Init(options));
 
-    server_.reset(new InProcessTestServer(std::move(server), location));
-    ASSERT_OK(server_->Start());
+    ASSERT_OK(Location::ForGrpcTls("localhost", server_->port(), &location_));
     ASSERT_OK(ConnectClient());
   }
 
-  void TearDown() {
-    if (server_) {
-      server_->Stop();
-    }
-  }
+  void TearDown() { ASSERT_OK(server_->Shutdown()); }
 
   Status ConnectClient() {
     auto options = FlightClientOptions();
     CertKeyPair root_cert;
     RETURN_NOT_OK(ExampleTlsCertificateRoot(&root_cert));
     options.tls_root_certs = root_cert.pem_cert;
-    return FlightClient::Connect(server_->location(), options, &client_);
+    return FlightClient::Connect(location_, options, &client_);
   }
 
  protected:
+  Location location_;
   std::unique_ptr<FlightClient> client_;
-  std::unique_ptr<InProcessTestServer> server_;
+  std::unique_ptr<FlightServerBase> server_;
 };
 
+// A server middleware that rejects all calls.
 class RejectServerMiddlewareFactory : public ServerMiddlewareFactory {
   Status StartCall(const CallInfo& info, const CallHeaders& incoming_headers,
                    std::shared_ptr<ServerMiddleware>* middleware) override {
@@ -590,12 +576,14 @@ class RejectServerMiddlewareFactory : public ServerMiddlewareFactory {
   }
 };
 
+// A server middleware that counts the number of successful and failed
+// calls.
 class CountingServerMiddleware : public ServerMiddleware {
  public:
   CountingServerMiddleware(std::atomic<int>* successful, std::atomic<int>* failed)
       : successful_(successful), failed_(failed) {}
-  void SendingHeaders(AddCallHeaders& outgoing_headers) {}
-  void CallCompleted(const Status& status) {
+  void SendingHeaders(AddCallHeaders* outgoing_headers) override {}
+  void CallCompleted(const Status& status) override {
     if (status.ok()) {
       ARROW_IGNORE_EXPR((*successful_)++);
     } else {
@@ -630,24 +618,27 @@ class TracingServerMiddlewareFactory : public ServerMiddlewareFactory {
 
   Status StartCall(const CallInfo& info, const CallHeaders& incoming_headers,
                    std::shared_ptr<ServerMiddleware>* middleware) override {
-    const auto& iter_pair = incoming_headers.GetHeaders("x-tracing-span-id");
+    const std::pair<HeaderIterator, HeaderIterator>& iter_pair =
+        incoming_headers.GetHeaders("x-tracing-span-id");
     if (iter_pair.first != iter_pair.second) {
-      const auto& value = (*iter_pair.first).second;
-      const std::string copy(value);
-      current_span_id = copy;
+      const util::string_view& value = (*iter_pair.first).second;
+      current_span_id = std::string(value);
     }
     return Status::OK();
   }
 };
 
+// A client middleware that adds a thread-local "request ID" to
+// outgoing calls as a header, and keeps track of the status of
+// completed calls. NOT thread-safe.
 class PropagatingClientMiddleware : public ClientMiddleware {
  public:
   explicit PropagatingClientMiddleware(std::atomic<int>* received_headers,
                                        std::vector<Status>* recorded_status)
       : received_headers_(received_headers), recorded_status_(recorded_status) {}
 
-  void SendingHeaders(AddCallHeaders& outgoing_headers) {
-    outgoing_headers.AddHeader("x-tracing-span-id", current_span_id);
+  void SendingHeaders(AddCallHeaders* outgoing_headers) {
+    outgoing_headers->AddHeader("x-tracing-span-id", current_span_id);
   }
 
   void ReceivedHeaders(const CallHeaders& incoming_headers) { (*received_headers_)++; }
@@ -702,61 +693,45 @@ class PropagatingTestServer : public FlightServerBase {
   std::unique_ptr<FlightClient> client_;
 };
 
-class TestServerMiddleware : public ::testing::Test {
+class TestRejectServerMiddleware : public ::testing::Test {
  public:
   void SetUp() {
-    Location location;
-    std::unique_ptr<FlightServerBase> server(new MetadataTestServer);
-
-    ASSERT_OK(Location::ForGrpcTcp("localhost", 0, &location));
-    FlightServerOptions options(location);
-    options.middleware.push_back(std::make_shared<RejectServerMiddlewareFactory>());
-    ASSERT_OK(server->Init(options));
-
-    Location real_location;
-    ASSERT_OK(Location::ForGrpcTcp("localhost", server->port(), &real_location));
-    server_.reset(new InProcessTestServer(std::move(server), real_location));
-    ASSERT_OK(server_->Start());
-    ASSERT_OK(ConnectClient());
+    ASSERT_OK(MakeServer<MetadataTestServer>(
+        &server_, &client_,
+        [](FlightServerOptions* options) {
+          options->middleware.push_back(
+              std::make_shared<RejectServerMiddlewareFactory>());
+          return Status::OK();
+        },
+        [](FlightClientOptions* options) { return Status::OK(); }));
   }
 
-  void TearDown() { server_->Stop(); }
-
-  Status ConnectClient() { return FlightClient::Connect(server_->location(), &client_); }
+  void TearDown() { ASSERT_OK(server_->Shutdown()); }
 
  protected:
   std::unique_ptr<FlightClient> client_;
-  std::unique_ptr<InProcessTestServer> server_;
+  std::unique_ptr<FlightServerBase> server_;
 };
 
 class TestCountingServerMiddleware : public ::testing::Test {
  public:
   void SetUp() {
-    Location location;
-    std::unique_ptr<FlightServerBase> server(new MetadataTestServer);
-
     request_counter_ = std::make_shared<CountingServerMiddlewareFactory>();
-
-    ASSERT_OK(Location::ForGrpcTcp("localhost", 0, &location));
-    FlightServerOptions options(location);
-    options.middleware.push_back(request_counter_);
-    ASSERT_OK(server->Init(options));
-
-    Location real_location;
-    ASSERT_OK(Location::ForGrpcTcp("localhost", server->port(), &real_location));
-    server_.reset(new InProcessTestServer(std::move(server), real_location));
-    ASSERT_OK(server_->Start());
-    ASSERT_OK(ConnectClient());
+    ASSERT_OK(MakeServer<MetadataTestServer>(
+        &server_, &client_,
+        [&](FlightServerOptions* options) {
+          options->middleware.push_back(request_counter_);
+          return Status::OK();
+        },
+        [](FlightClientOptions* options) { return Status::OK(); }));
   }
 
-  void TearDown() { server_->Stop(); }
-
-  Status ConnectClient() { return FlightClient::Connect(server_->location(), &client_); }
+  void TearDown() { ASSERT_OK(server_->Shutdown()); }
 
  protected:
-  std::unique_ptr<FlightClient> client_;
-  std::unique_ptr<InProcessTestServer> server_;
   std::shared_ptr<CountingServerMiddlewareFactory> request_counter_;
+  std::unique_ptr<FlightClient> client_;
+  std::unique_ptr<FlightServerBase> server_;
 };
 
 // Setup for this test is 2 servers
@@ -769,37 +744,33 @@ class TestCountingServerMiddleware : public ::testing::Test {
 class TestPropagatingMiddleware : public ::testing::Test {
  public:
   void SetUp() {
-    Location location;
-    ASSERT_OK(Location::ForGrpcTcp("localhost", 0, &location));
-
     server_middleware_ = std::make_shared<TracingServerMiddlewareFactory>();
     second_client_middleware_ = std::make_shared<PropagatingClientMiddlewareFactory>();
     client_middleware_ = std::make_shared<PropagatingClientMiddlewareFactory>();
 
-    second_server_ = arrow::internal::make_unique<ReportContextTestServer>();
-    FlightServerOptions second_server_options(location);
-    second_server_options.middleware.push_back(server_middleware_);
-    ASSERT_OK(second_server_->Init(second_server_options));
-
-    Location real_location;
-    ASSERT_OK(Location::ForGrpcTcp("localhost", second_server_->port(), &real_location));
     std::unique_ptr<FlightClient> server_client;
-    FlightClientOptions second_client_options{};
-    second_client_options.middleware.push_back(second_client_middleware_);
-    ASSERT_OK(
-        FlightClient::Connect(real_location, second_client_options, &server_client));
+    ASSERT_OK(MakeServer<ReportContextTestServer>(
+        &second_server_, &server_client,
+        [&](FlightServerOptions* options) {
+          options->middleware.push_back(server_middleware_);
+          return Status::OK();
+        },
+        [&](FlightClientOptions* options) {
+          options->middleware.push_back(second_client_middleware_);
+          return Status::OK();
+        }));
 
-    first_server_ =
-        arrow::internal::make_unique<PropagatingTestServer>(std::move(server_client));
-
-    FlightServerOptions first_server_options(location);
-    first_server_options.middleware.push_back(server_middleware_);
-    ASSERT_OK(first_server_->Init(first_server_options));
-
-    ASSERT_OK(Location::ForGrpcTcp("localhost", first_server_->port(), &real_location));
-    FlightClientOptions first_client_options{};
-    first_client_options.middleware.push_back(client_middleware_);
-    ASSERT_OK(FlightClient::Connect(real_location, first_client_options, &client_));
+    ASSERT_OK(MakeServer<PropagatingTestServer>(
+        &first_server_, &client_,
+        [&](FlightServerOptions* options) {
+          options->middleware.push_back(server_middleware_);
+          return Status::OK();
+        },
+        [&](FlightClientOptions* options) {
+          options->middleware.push_back(client_middleware_);
+          return Status::OK();
+        },
+        std::move(server_client)));
   }
 
   void ValidateStatus(const Status& status, const FlightMethod& method) {
@@ -1249,7 +1220,7 @@ TEST_F(TestTls, OverrideHostname) {
   CertKeyPair root_cert;
   ASSERT_OK(ExampleTlsCertificateRoot(&root_cert));
   client_options.tls_root_certs = root_cert.pem_cert;
-  ASSERT_OK(FlightClient::Connect(server_->location(), client_options, &client));
+  ASSERT_OK(FlightClient::Connect(location_, client_options, &client));
 
   FlightCallOptions options;
   options.timeout = TimeoutDuration{5.0};
@@ -1327,7 +1298,7 @@ TEST_F(TestMetadata, DoPutReadMetadata) {
   ASSERT_OK(writer->Close());
 }
 
-TEST_F(TestServerMiddleware, Rejected) {
+TEST_F(TestRejectServerMiddleware, Rejected) {
   std::unique_ptr<FlightInfo> info;
   const auto& status = client_->GetFlightInfo(FlightDescriptor{}, &info);
   ASSERT_RAISES(IOError, status);
@@ -1452,6 +1423,8 @@ TEST_F(TestPropagatingMiddleware, CallHeaders) {
 
   internal::GrpcCallHeaders incoming_headers(&headers);
 
+  // Make sure that we didn't implement HeaderIterator::Impl
+  // incorrectly
   ASSERT_EQ(incoming_headers.cbegin(), incoming_headers.cbegin());
   ASSERT_EQ(incoming_headers.cend(), incoming_headers.cend());
   ASSERT_EQ(2, incoming_headers.Count(header1));

cpp/src/arrow/flight/middleware.h

@@ -109,7 +109,7 @@ enum class FlightMethod : char {
 };
 
 /// \brief Information about an instance of a Flight RPC.
-class ARROW_FLIGHT_EXPORT CallInfo {
+struct ARROW_FLIGHT_EXPORT CallInfo {
  public:
   /// \brief The RPC method of this call.
   FlightMethod method;

cpp/src/arrow/flight/server.cc

@@ -71,7 +71,7 @@ namespace flight {
 #define RETURN_WITH_MIDDLEWARE(CONTEXT, STATUS) \
   do {                                          \
     const auto& __s = (STATUS);                 \
-    return CONTEXT.finish_request(__s);         \
+    return CONTEXT.FinishRequest(__s);          \
   } while (false)
 
 #define CHECK_ARG_NOT_NULL(CONTEXT, VAL, MESSAGE)                                      \
@@ -86,7 +86,7 @@ namespace flight {
   do {                                       \
     const auto& _s = (expr);                 \
     if (ARROW_PREDICT_FALSE(!_s.ok())) {     \
-      return CONTEXT.finish_request(_s);     \
+      return CONTEXT.FinishRequest(_s);      \
     }                                        \
   } while (false)
 
@@ -239,13 +239,13 @@ class GrpcServerCallContext : public ServerCallContext {
 
   // Helper method that runs interceptors given the result of an RPC,
   // then returns the final gRPC status to send to the client
-  grpc::Status finish_request(const grpc::Status& status) {
+  grpc::Status FinishRequest(const grpc::Status& status) {
     // Don't double-convert status - return the original one here
-    finish_request(internal::FromGrpcStatus(status));
+    FinishRequest(internal::FromGrpcStatus(status));
     return status;
   }
 
-  grpc::Status finish_request(const arrow::Status& status) {
+  grpc::Status FinishRequest(const arrow::Status& status) {
     for (const auto& instance : middleware_) {
       instance->CallCompleted(status);
     }
@@ -358,11 +358,11 @@ class FlightServiceImpl : public FlightService::Service {
       if (!result.ok()) {
         // Interceptor rejected call, end the request on all existing
         // interceptors
-        return flight_context.finish_request(result);
+        return flight_context.FinishRequest(result);
       }
-      if (instance) {
+      if (instance != nullptr) {
         flight_context.middleware_.push_back(instance);
-        instance->SendingHeaders(outgoing_headers);
+        instance->SendingHeaders(&outgoing_headers);
       }
     }
 
@@ -621,11 +621,7 @@ thread_local std::atomic<FlightServerBase::Impl*>
 #endif
 
 FlightServerOptions::FlightServerOptions(const Location& location_)
-    : location(location_),
-      auth_handler(nullptr),
-      tls_certificates(),
-      middleware(),
-      builder_hook() {}
+    : location(location_), auth_handler(nullptr) {}
 
 FlightServerOptions::~FlightServerOptions() = default;
 

cpp/src/arrow/flight/server_middleware.h

@@ -27,12 +27,11 @@
 #include "arrow/status.h"
 
 namespace arrow {
-
 namespace flight {
 
 /// \brief Server-side middleware for a call, instantiated per RPC.
 ///
-/// Middleware should be fast and must be infalliable: there is no way
+/// Middleware should be fast and must be infallible: there is no way
 /// to reject the call or report errors from the middleware instance.
 class ARROW_FLIGHT_EXPORT ServerMiddleware {
  public:
@@ -40,7 +39,7 @@ class ARROW_FLIGHT_EXPORT ServerMiddleware {
 
   /// \brief A callback before headers are sent. Extra headers can be
   /// added, but existing ones cannot be read.
-  virtual void SendingHeaders(AddCallHeaders& outgoing_headers) = 0;
+  virtual void SendingHeaders(AddCallHeaders* outgoing_headers) = 0;
 
   /// \brief A callback after the call has completed.
   virtual void CallCompleted(const Status& status) = 0;
@@ -65,7 +64,7 @@ class ARROW_FLIGHT_EXPORT ServerMiddlewareFactory {
   /// \param incoming_headers Headers sent by the client for this call.
   ///     Do not retain a reference to this object.
   /// \param[out] middleware The middleware instance for this call. If
-  ///     unset, will not add middleware to this call instance from
+  ///     null, no middleware will be added to this call instance from
   ///     this factory.
   /// \return Status A non-OK status will reject the call with the
   ///     given status. Middleware previously in the chain will have
@@ -76,5 +75,4 @@ class ARROW_FLIGHT_EXPORT ServerMiddlewareFactory {
 };
 
 }  // namespace flight
-
 }  // namespace arrow

cpp/src/arrow/flight/test_util.cc

@@ -132,25 +132,6 @@ bool TestServer::IsRunning() { return server_process_->running(); }
 
 int TestServer::port() const { return port_; }
 
-Status InProcessTestServer::Start() {
-  thread_ = std::thread([this]() { ARROW_EXPECT_OK(server_->Serve()); });
-  return Status::OK();
-}
-
-void InProcessTestServer::Stop() {
-  ARROW_CHECK_OK(server_->Shutdown());
-  thread_.join();
-}
-
-const Location& InProcessTestServer::location() const { return location_; }
-
-InProcessTestServer::~InProcessTestServer() {
-  // Make sure server shuts down properly
-  if (thread_.joinable()) {
-    Stop();
-  }
-}
-
 Status GetBatchForFlight(const Ticket& ticket, std::shared_ptr<RecordBatchReader>* out) {
   if (ticket.ticket == "ticket-ints-1") {
     BatchVector batches;

cpp/src/arrow/flight/test_util.h

@@ -66,22 +66,6 @@ class ARROW_FLIGHT_EXPORT TestServer {
   std::shared_ptr<::boost::process::child> server_process_;
 };
 
-class ARROW_FLIGHT_EXPORT InProcessTestServer {
- public:
-  explicit InProcessTestServer(std::unique_ptr<FlightServerBase> server,
-                               const Location& location)
-      : server_(std::move(server)), location_(location), thread_() {}
-  ~InProcessTestServer();
-  Status Start();
-  void Stop();
-  const Location& location() const;
-
- private:
-  std::unique_ptr<FlightServerBase> server_;
-  Location location_;
-  std::thread thread_;
-};
-
 /// \brief Create a simple Flight server for testing
 ARROW_FLIGHT_EXPORT
 std::unique_ptr<FlightServerBase> ExampleTestServer();

cpp/src/arrow/python/flight.cc

@@ -265,7 +265,7 @@ PyServerMiddleware::PyServerMiddleware(PyObject* middleware, Vtable vtable)
   middleware_.reset(middleware);
 }
 
-void PyServerMiddleware::SendingHeaders(arrow::flight::AddCallHeaders& outgoing_headers) {
+void PyServerMiddleware::SendingHeaders(arrow::flight::AddCallHeaders* outgoing_headers) {
   const Status& status = SafeCallIntoPython([&] {
     const Status status = vtable_.sending_headers(middleware_.obj(), outgoing_headers);
     RETURN_NOT_OK(CheckPyError());
@@ -316,7 +316,7 @@ PyClientMiddleware::PyClientMiddleware(PyObject* middleware, Vtable vtable)
   middleware_.reset(middleware);
 }
 
-void PyClientMiddleware::SendingHeaders(arrow::flight::AddCallHeaders& outgoing_headers) {
+void PyClientMiddleware::SendingHeaders(arrow::flight::AddCallHeaders* outgoing_headers) {
   const Status& status = SafeCallIntoPython([&] {
     const Status status = vtable_.sending_headers(middleware_.obj(), outgoing_headers);
     RETURN_NOT_OK(CheckPyError());

cpp/src/arrow/python/flight.h

@@ -205,7 +205,7 @@ class ARROW_PYTHON_EXPORT PyServerMiddlewareFactory
 class ARROW_PYTHON_EXPORT PyServerMiddleware : public arrow::flight::ServerMiddleware {
  public:
   typedef std::function<Status(PyObject*,
-                               arrow::flight::AddCallHeaders& outgoing_headers)>
+                               arrow::flight::AddCallHeaders* outgoing_headers)>
       SendingHeadersCallback;
   typedef std::function<Status(PyObject*, const Status& status)> CallCompletedCallback;
 
@@ -217,7 +217,7 @@ class ARROW_PYTHON_EXPORT PyServerMiddleware : public arrow::flight::ServerMiddl
   /// \brief Must only be called while holding the GIL.
   explicit PyServerMiddleware(PyObject* middleware, Vtable vtable);
 
-  void SendingHeaders(arrow::flight::AddCallHeaders& outgoing_headers) override;
+  void SendingHeaders(arrow::flight::AddCallHeaders* outgoing_headers) override;
   void CallCompleted(const Status& status) override;
 
  private:
@@ -248,7 +248,7 @@ class ARROW_PYTHON_EXPORT PyClientMiddlewareFactory
 class ARROW_PYTHON_EXPORT PyClientMiddleware : public arrow::flight::ClientMiddleware {
  public:
   typedef std::function<Status(PyObject*,
-                               arrow::flight::AddCallHeaders& outgoing_headers)>
+                               arrow::flight::AddCallHeaders* outgoing_headers)>
       SendingHeadersCallback;
   typedef std::function<Status(PyObject*,
                                const arrow::flight::CallHeaders& incoming_headers)>
@@ -264,7 +264,7 @@ class ARROW_PYTHON_EXPORT PyClientMiddleware : public arrow::flight::ClientMiddl
   /// \brief Must only be called while holding the GIL.
   explicit PyClientMiddleware(PyObject* factory, Vtable vtable);
 
-  void SendingHeaders(arrow::flight::AddCallHeaders& outgoing_headers) override;
+  void SendingHeaders(arrow::flight::AddCallHeaders* outgoing_headers) override;
   void ReceivedHeaders(const arrow::flight::CallHeaders& incoming_headers) override;
   void CallCompleted(const Status& status) override;
 

python/pyarrow/_flight.pyx

@@ -1645,7 +1645,7 @@ cdef CStatus _get_token(void* self, c_string* token) except *:
 
 
 cdef CStatus _middleware_sending_headers(
-        void* self, CAddCallHeaders& add_headers) except *:
+        void* self, CAddCallHeaders* add_headers) except *:
     """Callback for implementing middleware."""
     try:
         headers = (<object> self).sending_headers()
@@ -1855,7 +1855,7 @@ cdef class ClientMiddlewareFactory:
 cdef class ClientMiddleware:
     """Client-side middleware for a call, instantiated per RPC.
 
-    Methods here should be fast and must be infalliable: they should
+    Methods here should be fast and must be infallible: they should
     not raise exceptions or stall indefinitely.
 
     """
@@ -1866,7 +1866,7 @@ cdef class ClientMiddleware:
         Returns
         -------
         headers : dict
-            A dictionary of header values to add to the response, or
+            A dictionary of header values to add to the request, or
             None if no headers are to be added. The dictionary should
             have string keys and string or list-of-string values.
 
@@ -1875,10 +1875,12 @@ cdef class ClientMiddleware:
     def received_headers(self, headers):
         """A callback when headers are received.
 
+        The default implementation does nothing.
+
         Parameters
         ----------
         headers : dict
-            A dictionary of headers from the client. Keys are strings
+            A dictionary of headers from the server. Keys are strings
             and values are lists of bytes.
 
         """
@@ -1886,6 +1888,8 @@ cdef class ClientMiddleware:
     def call_completed(self, exception):
         """A callback when the call finishes.
 
+        The default implementation does nothing.
+
         Parameters
         ----------
         exception : ArrowException
@@ -2132,11 +2136,6 @@ cdef class FlightServerBase:
                 py_middleware,
                 start_call))
             c_options.get().middleware.push_back(c_middleware)
-            # for key, factory in middleware.items():
-            #     middleware_pair.first = tobytes(key)
-            #     middleware_pair.second.reset(
-            #         new CPyServerMiddlewareFactory(factory, start_call))
-            #     c_options.get().middleware.push_back(middleware_pair)
 
         vtable.list_flights = &_list_flights
         vtable.get_flight_info = &_get_flight_info

python/pyarrow/includes/libarrow_flight.pxd

@@ -353,7 +353,7 @@ ctypedef CStatus cb_client_authenticate(object, CClientAuthSender*,
                                         CClientAuthReader*)
 ctypedef CStatus cb_get_token(object, c_string*)
 
-ctypedef CStatus cb_middleware_sending_headers(object, CAddCallHeaders&)
+ctypedef CStatus cb_middleware_sending_headers(object, CAddCallHeaders*)
 ctypedef CStatus cb_middleware_call_completed(object, const CStatus&)
 ctypedef CStatus cb_client_middleware_received_headers(
     object, const CCallHeaders&)