GH-50282: [C++][FlightRPC] Refactor GRPC server and transport classes by Alex-PLACET · Pull Request #50407 · apache/arrow

Alex-PLACET · 2026-07-07T13:38:05Z

GitHub Issue: [C++][FlightRPC] Refactor GRPC server and transport classes #50282

The goal of the pullrequest is to refactor the FligtRPC code to prepare the async implementation. I moved code from grpc_server and transport_server to the _internal files which will be used in both sync and async server implmentations.
I also created dedicated function for specific responsabilities and reduce code complexity or duplication.

github-actions · 2026-07-07T13:38:41Z

Thanks for opening a pull request!

If this is not a minor PR. Could you open an issue for this pull request on GitHub? https://github.com/apache/arrow/issues/new/choose

Opening GitHub issues ahead of time contributes to the Openness of the Apache Arrow project.

Then could you also rename the pull request title in the following format?

GH-${GITHUB_ISSUE_ID}: [${COMPONENT}] ${SUMMARY}

or

MINOR: [${COMPONENT}] ${SUMMARY}

See also:

github-actions · 2026-07-07T13:40:28Z

⚠️ GitHub issue #50282 has been automatically assigned in GitHub to PR creator.

Alex-PLACET · 2026-07-08T09:13:15Z

@raulcd Can you launch the CI please

Alex-PLACET · 2026-07-09T11:18:43Z

@raulcd Can you run the CI again please?

Alex-PLACET · 2026-07-20T08:25:53Z

@raulcd Can you launch the CI please

Alex-PLACET · 2026-07-21T11:02:50Z

@raulcd Can you launch the CI please (I hope it will be the last time)

raulcd · 2026-07-22T11:20:32Z

@raulcd Can you launch the CI please (I hope it will be the last time)

Sorry @Alex-PLACET , I missed this comment. Just kicked it

raulcd · 2026-07-31T09:56:48Z

+    Protobuf_SOURCE=BUNDLED \
+    gRPC_SOURCE=BUNDLED \


We should probably bump ARROW_GRPC_REQUIRED_VERSION. I am pretty sure if Ubuntu 22.04 ships older GRPC than necessary, we will also have to update the Linux Package jobs (.deb) for old ubuntu (potentially also for old Red Hat?), I'll kick off Linux Packages.

maybe the label doesn't work well with draft, PRs, I am unsure why the Linux Package jobs did not trigger, I'll take a look.

raulcd · 2026-07-31T09:58:28Z

    ::grpc::ByteBuffer* buffer, arrow::flight::internal::FlightData* out);
+
+ARROW_FLIGHT_EXPORT
+bool IsRegisteredGrpcFlightDataMessage(


we should add some comments on why now we have to register/unregister FlightData messages. Are those necessary?

The point was I got some issue with protobuf and I though it was because of a mess with with FlightPayload. I think I introduce an issue at some point it fixed it later in my development. I just removed this mechanism as it is useless now.

Alex-PLACET · 2026-09-02T09:19:00Z

Most of the code was moved from transport_server.cc

Alex-PLACET · 2026-09-02T09:24:33Z

+  std::vector<std::shared_ptr<ServerMiddleware>> middleware_;
+  std::unordered_map<std::string, std::shared_ptr<ServerMiddleware>> middleware_map_;
+  CallHeaders incoming_headers_;
+};


Moved from https://github.com/apache/arrow/pull/50407/changes/BASE..20c2325e0b8481adf97aeac53b815f2509b9f301#diff-b3f03a3354d6ebe8a7212c9fd37b8c4333357552947460a941b39777baa28b17L115

Alex-PLACET · 2026-09-02T09:44:46Z

-    // Allow uploading messages of any length
-    builder.SetMaxReceiveMessageSize(-1);
-
-    const std::string scheme = uri.scheme();
    int port = 0;
-    if (scheme == kSchemeGrpc || scheme == kSchemeGrpcTcp || scheme == kSchemeGrpcTls) {
-      std::stringstream address;
-      address << arrow::util::UriEncodeHost(uri.host()) << ':' << uri.port_text();
-
-      std::shared_ptr<::grpc::ServerCredentials> creds;
-      if (scheme == kSchemeGrpcTls) {
-        ::grpc::SslServerCredentialsOptions ssl_options;
-        for (const auto& pair : options.tls_certificates) {
-          ssl_options.pem_key_cert_pairs.push_back({pair.pem_key, pair.pem_cert});
-        }
-        if (options.verify_client) {
-          ssl_options.client_certificate_request =
-              GRPC_SSL_REQUEST_AND_REQUIRE_CLIENT_CERTIFICATE_AND_VERIFY;
-        }
-        if (!options.root_certificates.empty()) {
-          ssl_options.pem_root_certs = options.root_certificates;
-        }
-        creds = ::grpc::SslServerCredentials(ssl_options);
-      } else {
-        creds = ::grpc::InsecureServerCredentials();
-      }
-
-      builder.AddListeningPort(address.str(), creds, &port);
-    } else if (scheme == kSchemeGrpcUnix) {
-      std::stringstream address;
-      address << "unix:" << uri.path();
-      builder.AddListeningPort(address.str(), ::grpc::InsecureServerCredentials());
-      location_ = options.location;
-    } else {
-      return Status::NotImplemented("Scheme is not supported: " + scheme);
-    }


Moved to a dedicated function: AddServerListeningPort

Alex-PLACET · 2026-09-02T09:45:30Z

+Status AddServerListeningPort(const FlightServerOptions& options,
+                              const arrow::util::Uri& uri, ::grpc::ServerBuilder* builder,
+                              Location* location, int* port) {
+  const std::string scheme = uri.scheme();
+  if (scheme == kSchemeGrpc || scheme == kSchemeGrpcTcp || scheme == kSchemeGrpcTls) {
+    std::stringstream address;
+    address << arrow::util::UriEncodeHost(uri.host()) << ':' << uri.port_text();
+
+    std::shared_ptr<::grpc::ServerCredentials> creds;
+    if (scheme == kSchemeGrpcTls) {
+      ::grpc::SslServerCredentialsOptions ssl_options;
+      for (const auto& pair : options.tls_certificates) {
+        ssl_options.pem_key_cert_pairs.push_back({pair.pem_key, pair.pem_cert});
+      }
+      if (options.verify_client) {
+        ssl_options.client_certificate_request =
+            GRPC_SSL_REQUEST_AND_REQUIRE_CLIENT_CERTIFICATE_AND_VERIFY;
+      }
+      if (!options.root_certificates.empty()) {
+        ssl_options.pem_root_certs = options.root_certificates;
+      }
+      creds = ::grpc::SslServerCredentials(ssl_options);
+    } else {
+      creds = ::grpc::InsecureServerCredentials();
+    }
+    builder->AddListeningPort(address.str(), creds, port);
+    return Status::OK();
+  }
+  if (scheme == kSchemeGrpcUnix) {
+    std::stringstream address;
+    address << "unix:" << uri.path();
+    builder->AddListeningPort(address.str(), ::grpc::InsecureServerCredentials());
+    *location = options.location;
+    return Status::OK();
+  }
+  return Status::NotImplemented("Scheme is not supported: " + scheme);
+}


Moved from grpc_server.cc

Alex-PLACET · 2026-09-02T09:59:07Z

 }

-namespace {
-class TransportIpcMessageReader : public ipc::MessageReader {


Moved to transport_server_internal.cc

Alex-PLACET · 2026-09-02T11:30:39Z

Most of the code comes from server.cc

Alex-PLACET · 2026-09-02T11:36:16Z

Most of the code moved to grpc_server_internal cpp and h

Copilot

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

pitrou · 2026-10-06T11:55:23Z

@lidavidm Would you like to take a look at this?

lidavidm

It generally seems fine to me.

lidavidm · 2026-10-07T07:49:53Z

+ARROW_FLIGHT_EXPORT
+Status SetServerLocationFromUri(const arrow::util::Uri& uri, int port,
+                                Location* location);


nit: this should return Result; it's also not really clear to me why it's factored out

lidavidm · 2026-10-07T07:51:21Z

+
+namespace arrow::flight::transport::grpc {
+
+Status AddServerListeningPort(const FlightServerOptions& options,


nit: it might make a little more sense for this to parse a URI and return the creds/location/address instead of just taking a builder, in case we want to unit test this later

lidavidm · 2026-10-07T07:53:00Z

+/// Provides shared helpers for constructing message readers/writers from a
+/// transport-level data stream, and for writing a FlightDataStream to a
+/// transport-level stream (used by DoGet/DoExchange).
+class ARROW_FLIGHT_EXPORT ServerTransportBase {


Does this really need to be a base class? It seems WriteDataStream could just be a helper function

lidavidm · 2026-10-07T07:53:26Z

+ARROW_FLIGHT_EXPORT
+int PortFromLocation(const Location& location);


Could this just be a method on Location?

lidavidm · 2026-10-07T07:53:35Z

+ARROW_FLIGHT_EXPORT
+arrow::Result<arrow::util::Uri> ParseLocationUri(const Location& location);


Possibly a method on Location? That would also help resolve the name, which I found a bit confusing

- Add Location::uri() and Location::port() and update FlightServerBase::port() to return arrow::Result<int>. - Remove ParseLocationUri/PortFromLocation and use Location methods instead. - Remove ServerTransportBase; store MemoryManager on ServerTransport and adjust ctor. - Make WriteDataStream a free function (transport_server_internal). - Introduce GrpcServerEndpoint and ParseServerEndpoint (address + credentials + location) and update gRPC server startup to use it. - Update related headers and implementations to match the new internal API.

Copilot

Copilot review overview

🟡 Changes recommended

Unresolved API, compilation, header-installation, gRPC macro, and signal-initialization issues block approval.

Review effort: Lite
Findings: 6

Open (6)

Update callers for the Result<int> port API · New
Restore fallback for missing gRPC version macros · New
Remove internal header dependency from public transport header · New
Include declarations for IPC statistics types · New
Initialize shutdown callback before signal handling · New
Include complete Uri definition in types.h · New

+#define GRPC_CPP_VERSION_CHECK(major, minor, patch)                             \
+  ((GRPC_CPP_VERSION_MAJOR > (major) ||                                         \
+    (GRPC_CPP_VERSION_MAJOR == (major) && GRPC_CPP_VERSION_MINOR > (minor)) ||  \
+    ((GRPC_CPP_VERSION_MAJOR == (major) && GRPC_CPP_VERSION_MINOR == (minor) && \
+      GRPC_CPP_VERSION_PATCH >= (patch)))))


 #include <memory>

 #include "arrow/flight/transport.h"
+#include "arrow/flight/transport_server_internal.h"


+#include "arrow/flight/platform.h"
+#include "arrow/flight/server.h"
+#include "arrow/flight/visibility.h"
+#include "arrow/result.h"
+#include "arrow/status.h"
+#include "arrow/util/io_util.h"
+#include "arrow/util/logging.h"
+#include "arrow/util/uri.h"


+    shutdown_ = [shutdown = std::forward<ShutdownFn>(shutdown)]() mutable {
+      return shutdown(nullptr);
+    };
+    shutdown_warning_ = shutdown_warning;


+  /// \brief Get the URI representation of this location.
+  ///
+  /// \return Arrow result with the URI.
+  arrow::Result<arrow::util::Uri> uri() const;


Copilot

🟡 Changes recommended

Unresolved build, API compatibility, linking, and signal-handling issues remain.

11 open findings

Install transport_server_internal.h with public headers · New
Guard callback include for supported gRPC versions · New
Define callback-context FinishRequest overload · New
Install transport_server_internal.h with Meson · New
Synchronize shutdown callback lifetime · New
Include complete Uri definition in types.h
Initialize shutdown callback before signal handling
Include declarations for IPC statistics types
Remove internal header dependency from public transport header
Restore fallback for missing gRPC version macros
Update callers for the Result<int> port API

Previously missed (1)

In code that hasn't changed since last review

Return an error when URI port is unavailable

cpp/src/arrow/flight/types.cc:849

Uri::port() returns -1 when the URI has no port, so this implementation returns a successful Result containing -1 for locations such as grpc+unix. That contradicts the new FlightServerBase::port() contract, which promises an error when the port cannot be determined; convert the no-port sentinel to an error here.

🧠 Review effort: Lite

    'server_tracing_middleware.cc',
    'transport.cc',
    'transport_server.cc',
+    'transport_server_internal.cc',


+
+#include <sstream>
+
+#include <grpcpp/support/server_callback.h>


+ARROW_FLIGHT_EXPORT
+::grpc::Status ToGrpcStatus(const Status& arrow_status,
+                            ::grpc::CallbackServerContext* ctx);


 #include <memory>

 #include "arrow/flight/transport.h"
+#include "arrow/flight/transport_server_internal.h"


+    running_instance_ = nullptr;
+    shutdown_ = nullptr;
+    shutdown_warning_ = nullptr;


Copilot

🟡 Changes recommended

Unresolved API, signal-handling, build-installation, and callback implementation issues block approval.

12 open findings

Serve error path leaves signal handlers overridden · New
Synchronize shutdown callback lifetime
Install transport_server_internal.h with Meson
Define callback-context FinishRequest overload
Guard callback include for supported gRPC versions
Install transport_server_internal.h with public headers
Include complete Uri definition in types.h
Initialize shutdown callback before signal handling
Include declarations for IPC statistics types
Remove internal header dependency from public transport header
Restore fallback for missing gRPC version macros
Update callers for the Result<int> port API

🧠 Review effort: Lite

 Status FlightServerBase::Serve() {
-  if (!impl_->transport_) {
-    return Status::UnknownError("Server did not start properly");
-  }
-  impl_->got_signal_ = 0;
-  impl_->old_signal_handlers_.clear();
-  impl_->running_instance_ = impl_.get();
-
-  ServerSignalHandler signal_handler;
-  ARROW_ASSIGN_OR_RAISE(impl_->self_pipe_, signal_handler.Init(&Impl::WaitForSignals));
-  // Override existing signal handlers with our own handler so as to stop the server.
-  for (size_t i = 0; i < impl_->signals_.size(); ++i) {
-    int signum = impl_->signals_[i];
-    SignalHandler new_handler(&Impl::HandleSignal), old_handler;
-    ARROW_ASSIGN_OR_RAISE(old_handler, SetSignalHandler(signum, new_handler));
-    impl_->old_signal_handlers_.push_back(std::move(old_handler));
-  }
-
-  RETURN_NOT_OK(impl_->transport_->Wait());
-  impl_->running_instance_ = nullptr;
-
-  // Restore signal handlers
-  for (size_t i = 0; i < impl_->signals_.size(); ++i) {
-    RETURN_NOT_OK(
-        SetSignalHandler(impl_->signals_[i], impl_->old_signal_handlers_[i]).status());
-  }
-  return Status::OK();
+  return impl_->signal_state_.Serve(
+      [this]() -> Status {
+        if (!impl_->transport_) {
+          return Status::UnknownError("Server did not start properly");
+        }
+        return impl_->transport_->Wait();


This reverts commit bcf31a4.

Copilot

🟡 Changes recommended

Unresolved build, lifecycle, and concurrency issues must be addressed before approval.

13 open findings

Shutdown clears active server signal target on uninitialized object · New
Inconsistent gRPC version checks cause cross-TU link failures · New
Serve error path leaves signal handlers overridden
Synchronize shutdown callback lifetime
Install transport_server_internal.h with Meson
Define callback-context FinishRequest overload
Guard callback include for supported gRPC versions
Install transport_server_internal.h with public headers
Include complete Uri definition in types.h
Initialize shutdown callback before signal handling
Include declarations for IPC statistics types
Remove internal header dependency from public transport header
Restore fallback for missing gRPC version macros

1 resolved since last review

Update callers for the Result<int> port API

🧠 Review effort: Lite

Give feedback about Copilot approvals in this survey to enter a drawing for a $150 gift card.

-    return impl_->transport_->Shutdown(*deadline);
-  }
-  return impl_->transport_->Shutdown();
+  return impl_->signal_state_.Shutdown(


+#define GRPC_CPP_VERSION_CHECK(major, minor, patch)                             \
+  ((GRPC_CPP_VERSION_MAJOR > (major) ||                                         \
+    (GRPC_CPP_VERSION_MAJOR == (major) && GRPC_CPP_VERSION_MINOR > (minor)) ||  \
+    ((GRPC_CPP_VERSION_MAJOR == (major) && GRPC_CPP_VERSION_MINOR == (minor) && \
+      GRPC_CPP_VERSION_PATCH >= (patch)))))


github-actions Bot added the awaiting review Awaiting review label Jul 7, 2026

github-actions Bot added Component: C++ Component: FlightRPC labels Jul 7, 2026

Alex-PLACET changed the title ~~GH-#50282: [C++][FlightRPC] Refactor GRPC server and transport classes~~ GH-50282: [C++][FlightRPC] Refactor GRPC server and transport classes Jul 7, 2026

Alex-PLACET force-pushed the refactor/internal-transport branch 2 times, most recently from 560bd8d to 20c2325 Compare July 23, 2026 11:41

raulcd reviewed Jul 31, 2026

View reviewed changes

raulcd added the CI: Extra: Package: Linux Run extra Linux Packages CI label Jul 31, 2026

github-actions Bot added awaiting changes Awaiting changes and removed awaiting review Awaiting review labels Jul 31, 2026

Alex-PLACET commented Sep 2, 2026

View reviewed changes

github-actions Bot added awaiting change review Awaiting change review and removed awaiting changes Awaiting changes labels Sep 2, 2026

Alex-PLACET commented Sep 2, 2026

View reviewed changes

Comment thread cpp/src/arrow/flight/transport_server_internal.h

Alex-PLACET Sep 2, 2026

Copy link
Copy Markdown

Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Most of the code comes from server.cc

Alex-PLACET commented Sep 2, 2026

View reviewed changes

Alex-PLACET marked this pull request as ready for review September 17, 2026 14:40

Copilot AI lite review requested due to automatic review settings September 17, 2026 14:40

Alex-PLACET requested review from assignUser and lidavidm as code owners September 17, 2026 14:40

raulcd added the CI: Extra: R Run extra R CI label Sep 18, 2026

Try fix compilation of ubuntu 22.04 using bundled nlohmann_json_SOURCE

71b1b11

Copilot AI review requested due to automatic review settings September 18, 2026 08:37

Copilot AI reviewed Sep 18, 2026

Alex-PLACET requested a review from raulcd October 6, 2026 11:27

lidavidm reviewed Oct 7, 2026

View reviewed changes

github-actions Bot added awaiting review Awaiting review awaiting changes Awaiting changes and removed awaiting change review Awaiting change review labels Oct 7, 2026

Copilot AI lite review requested due to automatic review settings October 7, 2026 14:08

github-actions Bot added awaiting review Awaiting review and removed awaiting review Awaiting review awaiting changes Awaiting changes labels Oct 7, 2026

Copilot started reviewing on behalf of Alex-PLACET October 7, 2026 14:11 View session

Copilot AI reviewed Oct 7, 2026

View reviewed changes

fix

bcf31a4

Copilot AI lite review requested due to automatic review settings October 7, 2026 14:48

Copilot started reviewing on behalf of Alex-PLACET October 7, 2026 14:49 View session

Copilot AI reviewed Oct 7, 2026

View reviewed changes

fix

0ffdd13

Copilot AI lite review requested due to automatic review settings October 7, 2026 15:21

Copilot started reviewing on behalf of Alex-PLACET October 7, 2026 15:21 View session

Copilot AI reviewed Oct 7, 2026

View reviewed changes

Alex-PLACET added 2 commits October 8, 2026 09:05

Revert "fix"

e0e927c

This reverts commit bcf31a4.

wip

a9b43d2

Copilot AI lite review requested due to automatic review settings October 8, 2026 07:06

Copilot started reviewing on behalf of Alex-PLACET October 8, 2026 07:06 View session

Copilot AI reviewed Oct 8, 2026

View reviewed changes


		namespace arrow::flight::transport::grpc {

		Status AddServerListeningPort(const FlightServerOptions& options,

		ARROW_FLIGHT_EXPORT
		int PortFromLocation(const Location& location);

		ARROW_FLIGHT_EXPORT
		arrow::Result<arrow::util::Uri> ParseLocationUri(const Location& location);


		#include <sstream>

		#include <grpcpp/support/server_callback.h>

Conversation

Alex-PLACET commented Jul 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions Bot commented Jul 7, 2026

Uh oh!

github-actions Bot commented Jul 7, 2026

Uh oh!

Alex-PLACET commented Jul 8, 2026

Uh oh!

Alex-PLACET commented Jul 9, 2026

Uh oh!

Alex-PLACET commented Jul 20, 2026

Uh oh!

Alex-PLACET commented Jul 21, 2026

Uh oh!

raulcd commented Jul 22, 2026

Uh oh!

raulcd Jul 31, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Alex-PLACET Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Uh oh!

pitrou commented Oct 6, 2026

Uh oh!

lidavidm left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Copilot review overview

🟡 Changes recommended

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Alex-PLACET commented Jul 7, 2026 •

edited

Loading

raulcd Jul 31, 2026 •

edited

Loading

Alex-PLACET Sep 2, 2026 •

edited

Loading