Uh oh!
There was an error while loading. Please reload this page.
GH-50282: [C++][FlightRPC] Refactor GRPC server and transport classes - #50407
GH-50282: [C++][FlightRPC] Refactor GRPC server and transport classes#50407Alex-PLACET wants to merge 15 commits into
Conversation
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? or See also: |
Alex-PLACET
commented
Jul 8, 2026
@raulcd Can you launch the CI please |
Alex-PLACET
commented
Jul 9, 2026
@raulcd Can you run the CI again please? |
Alex-PLACET
commented
Jul 20, 2026
@raulcd Can you launch the CI please |
Alex-PLACET
commented
Jul 21, 2026
@raulcd Can you launch the CI please (I hope it will be the last time) |
raulcd
commented
Jul 22, 2026
Sorry @Alex-PLACET , I missed this comment. Just kicked it |
f3c0c73 to
560bd8dCompare560bd8d to
20c2325Compare| Protobuf_SOURCE=BUNDLED \ | ||
| gRPC_SOURCE=BUNDLED \ |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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.
| ::grpc::ByteBuffer* buffer, arrow::flight::internal::FlightData* out); | ||
| ARROW_FLIGHT_EXPORT | ||
| bool IsRegisteredGrpcFlightDataMessage( |
There was a problem hiding this comment.
we should add some comments on why now we have to register/unregister FlightData messages. Are those necessary?
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
Most of the code was moved from transport_server.cc
| std::vector<std::shared_ptr<ServerMiddleware>> middleware_; | ||
| std::unordered_map<std::string, std::shared_ptr<ServerMiddleware>> middleware_map_; | ||
| CallHeaders incoming_headers_; | ||
| }; |
There was a problem hiding this comment.
| // 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); | ||
| } |
There was a problem hiding this comment.
Moved to a dedicated function: AddServerListeningPort
| 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); | ||
| } |
| } | ||
| namespace { | ||
| class TransportIpcMessageReader : public ipc::MessageReader { |
There was a problem hiding this comment.
Moved to transport_server_internal.cc
There was a problem hiding this comment.
Most of the code comes from server.cc
There was a problem hiding this comment.
Most of the code moved to grpc_server_internal cpp and h
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.