GH-41947: [Java] Support catalog in JDBC driver with session options

[stevelorddremio]

Rationale for this change

See Issue #41947

What changes are included in this PR?

Are these changes tested?

Yes

Are there any user-facing changes?

Introductiona of an optional catalog query parameter in the JDBC url string.

• GitHub Issue: [Java] Add support for catalog as a JDBC query parameter #41947

[github-actions]

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

[vibhatha]

Overall LGTM!

[stevelorddremio]

All updates have been incorporated.
Ubuntu JDK 17 build fails. I think this will pass if re-triggered.

[lidavidm]

@stevelorddremio thanks, do you mind rebasing/running the new formatter?

[stevelorddremio]

I think AMD64 Windows Server 2022 Java JDK 11 will pass if re-triggered.

[lidavidm]

Sorry for the back and forth. This looks good to me now!

[conbench-apache-arrow]

After merging your PR, Conbench analyzed the 6 benchmarking runs that have been run so far on merge-commit cecd771.

There was 1 benchmark result indicating a performance regression:

• Commit Run on test-mac-arm at 2024-06-14 10:17:26Z
  • tpch (R) with engine=arrow, format=native, language=R, memory_map=False, query_id=TPCH-21, scale_factor=1

The full Conbench report has more details. It also includes information about 16 possible false positives for unstable benchmarks that are known to sometimes produce them.

[scruz-denodo]

Hi, I was testing this change and I am facing a problem when the java.sql.Connection#close method of the JDBC connection is called.

I am receiving the following error invoking the method.

Exception in thread "main" java.sql.SQLException: UNKNOWN: Uncaught exception in the SynchronizationContext. Re-thrown.
	at org.apache.calcite.avatica.Helper.createException(Helper.java:56)
	at org.apache.calcite.avatica.Helper.createException(Helper.java:41)
	at org.apache.arrow.driver.jdbc.ArrowFlightConnection.close(ArrowFlightConnection.java:187)
	at com.arrowflight.test.Main.test7(Main.java:249)
	at com.arrowflight.test.Main.main(Main.java:27)
Caused by: io.grpc.StatusRuntimeException: UNKNOWN: Uncaught exception in the SynchronizationContext. Re-thrown.
	at io.grpc.Status.asRuntimeException(Status.java:525)
	at io.grpc.internal.RetriableStream$1.uncaughtException(RetriableStream.java:75)
	at io.grpc.SynchronizationContext.drain(SynchronizationContext.java:96)
	at io.grpc.SynchronizationContext.execute(SynchronizationContext.java:126)
	at io.grpc.internal.RetriableStream.safeCloseMasterListener(RetriableStream.java:838)
	at io.grpc.internal.RetriableStream.cancel(RetriableStream.java:531)
	at io.grpc.internal.RetriableStream.start(RetriableStream.java:393)
	at io.grpc.internal.ClientCallImpl.startInternal(ClientCallImpl.java:285)
	at io.grpc.internal.ClientCallImpl.start(ClientCallImpl.java:184)
	at io.grpc.ForwardingClientCall.start(ForwardingClientCall.java:32)
	at org.apache.arrow.flight.grpc.ClientInterceptorAdapter$FlightClientCall.start(ClientInterceptorAdapter.java:142)
	at io.grpc.ForwardingClientCall.start(ForwardingClientCall.java:32)
	at io.grpc.stub.MetadataUtils$HeaderAttachingClientInterceptor$HeaderAttachingClientCall.start(MetadataUtils.java:75)
	at io.grpc.stub.ClientCalls.startCall(ClientCalls.java:335)
	at io.grpc.stub.ClientCalls.asyncUnaryRequestCall(ClientCalls.java:311)
	at io.grpc.stub.ClientCalls.blockingServerStreamingCall(ClientCalls.java:210)
	at org.apache.arrow.flight.impl.FlightServiceGrpc$FlightServiceBlockingStub.doAction(FlightServiceGrpc.java:874)
	at org.apache.arrow.flight.FlightClient.doAction(FlightClient.java:168)
	at org.apache.arrow.flight.FlightClient.closeSession(FlightClient.java:686)
	at org.apache.arrow.flight.sql.FlightSqlClient.closeSession(FlightSqlClient.java:998)
	at org.apache.arrow.driver.jdbc.client.ArrowFlightSqlClientHandler.close(ArrowFlightSqlClientHandler.java:224)
	at org.apache.arrow.util.AutoCloseables.close(AutoCloseables.java:97)
	at org.apache.arrow.util.AutoCloseables.close(AutoCloseables.java:75)
	at org.apache.arrow.driver.jdbc.ArrowFlightConnection.close(ArrowFlightConnection.java:181)
	... 2 more
Caused by: java.util.concurrent.RejectedExecutionException: Task io.grpc.internal.SerializingExecutor@6ecdbab8 rejected from java.util.concurrent.ThreadPoolExecutor@3dd4a6fa[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 25]
	at java.base/java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2055)
	at java.base/java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:825)
	at java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1355)
	at io.grpc.internal.SerializingExecutor.schedule(SerializingExecutor.java:102)
	at io.grpc.internal.SerializingExecutor.execute(SerializingExecutor.java:95)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.closedInternal(ClientCallImpl.java:736)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.closed(ClientCallImpl.java:680)
	at io.grpc.internal.RetriableStream$4.run(RetriableStream.java:843)
	at io.grpc.SynchronizationContext.drain(SynchronizationContext.java:94)
	... 23 more

I think the reason is the following. Checking this code:

arrow/java/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightConnection.java

Lines 174 to 189 in 6a2e19a

	public void close() throws SQLException {
	clientHandler.close();
	if (executorService != null) {
	executorService.shutdown();
	}
	try {
	AutoCloseables.close(clientHandler);
	allocator.getChildAllocators().forEach(AutoCloseables::closeNoChecked);
	AutoCloseables.close(allocator);
	super.close();
	} catch (final Exception e) {
	throw AvaticaConnection.HELPER.createException(e.getMessage(), e);
	}
	}

• close method invokes clientHandler.close(); at line 175. Here is where now is invoked the closeSession method at line 224.
  

  arrow/java/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/ArrowFlightSqlClientHandler.java

  Lines 222 to 225 in 6a2e19a

  	public void close() throws SQLException {
  	if (catalog.isPresent()) {
  	sqlClient.closeSession(new CloseSessionRequest(), getOptions());
  	}

• after that again on ArrowFlightConnection.java, the executorService is closed at like 177
• and the clientHandler.close(); is called again at line 181.
  • with this call, the closeSession method is invoked again. And it is here when the exception is thrown. Because the executorService was already shutdown.

arrow/java/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightConnection.java

Lines 174 to 189 in 6a2e19a

	public void close() throws SQLException {
	clientHandler.close();
	if (executorService != null) {
	executorService.shutdown();
	}
	try {
	AutoCloseables.close(clientHandler);
	allocator.getChildAllocators().forEach(AutoCloseables::closeNoChecked);
	AutoCloseables.close(allocator);
	super.close();
	} catch (final Exception e) {
	throw AvaticaConnection.HELPER.createException(e.getMessage(), e);
	}
	}

I think that line line 181 from org.apache.arrow.driver.jdbc.ArrowFlightConnection#close should be remove for avoding the problem. The close is already done at 175.

Also, I would remove the is if:

arrow/java/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/ArrowFlightSqlClientHandler.java

Lines 222 to 225 in 6a2e19a

	public void close() throws SQLException {
	if (catalog.isPresent()) {
	sqlClient.closeSession(new CloseSessionRequest(), getOptions());
	}

why not invoking the closeSession when the catalog was not set?

[scruz-denodo]

Please, let me know if I have to open a new issue with this problem.

I put here as first step because this version 17 is being currently published.

[lidavidm]

Please open a new issue

[vibhatha]

@github-actions crossbow submit -g java

[github-actions]

Revision: 9a43834

Submitted crossbow builds: ursacomputing/crossbow @ actions-f7d1d505f9

Task	Status
java-jars
test-conda-python-3.10-spark-v3.5.0
test-conda-python-3.11-spark-master
test-conda-python-3.8-spark-v3.5.0
verify-rc-source-java-linux-almalinux-8-amd64
verify-rc-source-java-linux-conda-latest-amd64
verify-rc-source-java-linux-ubuntu-20.04-amd64
verify-rc-source-java-linux-ubuntu-22.04-amd64
verify-rc-source-java-macos-amd64