Copilot commented on code in PR #719: URL: https://github.com/apache/doris-flink-connector/pull/719#discussion_r4479325314
########## flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisFlightSession.java: ########## @@ -0,0 +1,135 @@ +// 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. + +package org.apache.doris.flink.source.reader; + +import org.apache.arrow.adbc.core.AdbcConnection; +import org.apache.arrow.adbc.core.AdbcDatabase; +import org.apache.arrow.adbc.core.AdbcException; +import org.apache.arrow.adbc.driver.flightsql.FlightSqlClientWithCallOptions; +import org.apache.arrow.adbc.driver.flightsql.FlightSqlConnection; +import org.apache.arrow.flight.CallOption; +import org.apache.arrow.flight.CallOptions; +import org.apache.arrow.flight.CloseSessionRequest; +import org.apache.arrow.flight.CloseSessionResult; +import org.apache.arrow.flight.sql.FlightSqlClient; +import org.apache.doris.flink.exception.DorisRuntimeException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.lang.reflect.Field; +import java.util.Arrays; +import java.util.Objects; +import java.util.concurrent.TimeUnit; + +/** Closes the server session without taking ownership of the ADBC connection. */ +final class DorisFlightSession implements AutoCloseable { + private static final Logger LOG = LoggerFactory.getLogger(DorisFlightSession.class); + + private final AdbcConnection connection; + private final FlightSqlClient client; + private final CallOption[] callOptions; + private boolean closed; + + private DorisFlightSession( + AdbcConnection connection, FlightSqlClient client, CallOption[] callOptions) { + this.connection = connection; + this.client = client; + this.callOptions = Arrays.copyOf(callOptions, callOptions.length + 1); + this.callOptions[callOptions.length] = CallOptions.timeout(5, TimeUnit.SECONDS); + } + Review Comment: Appending a timeout `CallOption` assumes Arrow Flight will honor the last timeout option if multiple deadlines/timeouts exist in `connectionOptions`. To make the 5s bound unambiguous, consider filtering/removing any existing timeout/deadline-related options from `callOptions` before appending the new timeout for `CloseSession`. ########## flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/reader/DorisFlightSessionTest.java: ########## @@ -0,0 +1,218 @@ +// 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. + +package org.apache.doris.flink.source.reader; + +import org.apache.arrow.adbc.core.AdbcConnection; +import org.apache.arrow.adbc.core.AdbcDatabase; +import org.apache.arrow.adbc.core.AdbcDriver; +import org.apache.arrow.adbc.driver.flightsql.FlightSqlDriver; +import org.apache.arrow.flight.CallHeaders; +import org.apache.arrow.flight.CallStatus; +import org.apache.arrow.flight.CloseSessionRequest; +import org.apache.arrow.flight.CloseSessionResult; +import org.apache.arrow.flight.FlightServer; +import org.apache.arrow.flight.Location; +import org.apache.arrow.flight.auth2.CallHeaderAuthenticator; +import org.apache.arrow.flight.sql.NoOpFlightSqlProducer; +import org.apache.arrow.memory.RootAllocator; +import org.apache.doris.flink.exception.DorisRuntimeException; +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class DorisFlightSessionTest { + @Test + void closesTheSessionSharedWithAdbcOnlyOnce() throws Exception { + SessionProducer producer = new SessionProducer(); + try (RootAllocator allocator = new RootAllocator(); + FlightServer server = server(allocator, producer)) { + Map<String, Object> parameters = parameters(server); + try (AdbcDatabase database = new FlightSqlDriver(allocator).open(parameters)) { + DorisFlightSession session = DorisFlightSession.open(database); + try (AdbcConnection connection = session.getConnection(); + DorisFlightSession closingSession = session) { + assertThat(producer.opens.get()).isEqualTo(1); + assertThat(parameters) + .containsEntry(AdbcDriver.PARAM_USERNAME.getKey(), "root") + .containsEntry(AdbcDriver.PARAM_PASSWORD.getKey(), "password"); + closingSession.closeSession(); + closingSession.closeSession(); + assertThat(producer.closes.get()).isEqualTo(1); + assertThat(producer.bearerCalls.get()).isEqualTo(1); + assertThat(producer.active.get()).isZero(); + } + } + assertThat(producer.closes.get()).isEqualTo(1); + } + } + + @Test + void releasesAdbcResourcesEvenWhenCloseSessionFails() throws Exception { + SessionProducer producer = new SessionProducer(); + producer.rejectClose = true; + try (RootAllocator allocator = new RootAllocator(); + FlightServer server = server(allocator, producer); + AdbcDatabase database = new FlightSqlDriver(allocator).open(parameters(server))) { + DorisFlightSession session = DorisFlightSession.open(database); + assertThatCode(() -> DorisFlightValueReader.closeAll(session, session.getConnection())) + .doesNotThrowAnyException(); + session.close(); + assertThat(producer.closes.get()).isEqualTo(1); + } + } + + @Test + void boundsUnresponsiveSessionCloseAndContinuesCleanup() throws Exception { + SessionProducer producer = new SessionProducer(); + producer.hangClose = true; + try (RootAllocator allocator = new RootAllocator(); + FlightServer server = server(allocator, producer); + AdbcDatabase database = new FlightSqlDriver(allocator).open(parameters(server))) { + DorisFlightSession session = DorisFlightSession.open(database); + long start = System.nanoTime(); + assertThatCode(() -> DorisFlightValueReader.closeAll(session, session.getConnection())) + .doesNotThrowAnyException(); + assertThat(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)) + .isBetween(4000L, 10000L); Review Comment: This time-window assertion is prone to CI flakiness (scheduler stalls, GC pauses, slow networking) and can fail even if the 5s deadline is correctly applied. Prefer asserting an upper bound via JUnit timeout utilities (e.g., `assertTimeout`/`assertTimeoutPreemptively`) or significantly widen the upper bound while still validating it completes within a reasonable ceiling. ########## flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisFlightSession.java: ########## @@ -0,0 +1,135 @@ +// 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. + +package org.apache.doris.flink.source.reader; + +import org.apache.arrow.adbc.core.AdbcConnection; +import org.apache.arrow.adbc.core.AdbcDatabase; +import org.apache.arrow.adbc.core.AdbcException; +import org.apache.arrow.adbc.driver.flightsql.FlightSqlClientWithCallOptions; +import org.apache.arrow.adbc.driver.flightsql.FlightSqlConnection; +import org.apache.arrow.flight.CallOption; +import org.apache.arrow.flight.CallOptions; +import org.apache.arrow.flight.CloseSessionRequest; +import org.apache.arrow.flight.CloseSessionResult; +import org.apache.arrow.flight.sql.FlightSqlClient; +import org.apache.doris.flink.exception.DorisRuntimeException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.lang.reflect.Field; +import java.util.Arrays; +import java.util.Objects; +import java.util.concurrent.TimeUnit; + +/** Closes the server session without taking ownership of the ADBC connection. */ +final class DorisFlightSession implements AutoCloseable { + private static final Logger LOG = LoggerFactory.getLogger(DorisFlightSession.class); + + private final AdbcConnection connection; + private final FlightSqlClient client; + private final CallOption[] callOptions; + private boolean closed; + + private DorisFlightSession( + AdbcConnection connection, FlightSqlClient client, CallOption[] callOptions) { + this.connection = connection; + this.client = client; + this.callOptions = Arrays.copyOf(callOptions, callOptions.length + 1); + this.callOptions[callOptions.length] = CallOptions.timeout(5, TimeUnit.SECONDS); + } + + static DorisFlightSession open(AdbcDatabase database) throws AdbcException { + // shortcut: ADBC 0.14.0 hides the client; remove once its close() sends CloseSession. + Field connectionClient = + field(FlightSqlConnection.class, "client", FlightSqlClientWithCallOptions.class); + Field flightClient = + field(FlightSqlClientWithCallOptions.class, "client", FlightSqlClient.class); + Field connectionOptions = + field( + FlightSqlClientWithCallOptions.class, + "connectionOptions", + CallOption[].class); Review Comment: These reflective `Field` lookups run on every `open(...)`, which can add overhead when many splits repeatedly create sessions. Consider caching these `Field` instances as `private static final` members (initialized once) so per-session work is limited to `Field.get(...)` calls and validation happens once per classload. ########## flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisFlightSession.java: ########## @@ -0,0 +1,135 @@ +// 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. + +package org.apache.doris.flink.source.reader; + +import org.apache.arrow.adbc.core.AdbcConnection; +import org.apache.arrow.adbc.core.AdbcDatabase; +import org.apache.arrow.adbc.core.AdbcException; +import org.apache.arrow.adbc.driver.flightsql.FlightSqlClientWithCallOptions; +import org.apache.arrow.adbc.driver.flightsql.FlightSqlConnection; +import org.apache.arrow.flight.CallOption; +import org.apache.arrow.flight.CallOptions; +import org.apache.arrow.flight.CloseSessionRequest; +import org.apache.arrow.flight.CloseSessionResult; +import org.apache.arrow.flight.sql.FlightSqlClient; +import org.apache.doris.flink.exception.DorisRuntimeException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.lang.reflect.Field; +import java.util.Arrays; +import java.util.Objects; +import java.util.concurrent.TimeUnit; + +/** Closes the server session without taking ownership of the ADBC connection. */ +final class DorisFlightSession implements AutoCloseable { + private static final Logger LOG = LoggerFactory.getLogger(DorisFlightSession.class); + + private final AdbcConnection connection; + private final FlightSqlClient client; + private final CallOption[] callOptions; + private boolean closed; + + private DorisFlightSession( + AdbcConnection connection, FlightSqlClient client, CallOption[] callOptions) { + this.connection = connection; + this.client = client; + this.callOptions = Arrays.copyOf(callOptions, callOptions.length + 1); + this.callOptions[callOptions.length] = CallOptions.timeout(5, TimeUnit.SECONDS); + } + + static DorisFlightSession open(AdbcDatabase database) throws AdbcException { + // shortcut: ADBC 0.14.0 hides the client; remove once its close() sends CloseSession. + Field connectionClient = + field(FlightSqlConnection.class, "client", FlightSqlClientWithCallOptions.class); + Field flightClient = + field(FlightSqlClientWithCallOptions.class, "client", FlightSqlClient.class); + Field connectionOptions = + field( + FlightSqlClientWithCallOptions.class, + "connectionOptions", + CallOption[].class); + + // Validate the driver layout before authentication creates a server session. + AdbcConnection connection = database.connect(); + try { + Object wrapper = connectionClient.get(connection); + FlightSqlClient client = + Objects.requireNonNull((FlightSqlClient) flightClient.get(wrapper)); + CallOption[] options = + Objects.requireNonNull((CallOption[]) connectionOptions.get(wrapper)); + return new DorisFlightSession(connection, client, options); + } catch (ReflectiveOperationException | RuntimeException failure) { + try { + connection.close(); + } catch (Exception closeFailure) { + failure.addSuppressed(closeFailure); + } + throw new DorisRuntimeException("Cannot bind ADBC Flight SQL session", failure); + } + } + + private static Field field(Class<?> owner, String name, Class<?> type) { + try { + Field field = owner.getDeclaredField(name); + if (field.getType() != type) { + throw new NoSuchFieldException( + owner.getSimpleName() + "." + name + " has changed type"); + } + field.setAccessible(true); + return field; + } catch (ReflectiveOperationException | RuntimeException failure) { + throw new DorisRuntimeException( + "Unsupported ADBC Flight SQL driver: cannot access session for CloseSession", + failure); + } + } + + AdbcConnection getConnection() { + return connection; + } + + public synchronized void closeSession() { + if (closed) { + return; + } + closed = true; + // Remote session cleanup must not turn a completed read into a failure. + try { + CloseSessionResult.Status status = + client.closeSession(new CloseSessionRequest(), callOptions).getStatus(); Review Comment: Setting `closed = true` before attempting the RPC means any transient failure (timeout, temporary transport error) prevents any later retry attempt, which can still leave the remote session leaked even if subsequent cleanup calls occur. Optional: track 'attempted' vs 'successfully closed' (or only flip `closed` after reaching a terminal status like `CLOSED`/`CLOSING`) so a later close path can retry once without risking repeated attempts. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
