stankiewicz opened a new issue, #41157:
URL: https://github.com/apache/beam/issues/41157
### What happened?
When reading from Solace using `SolaceIO.Read`, if the underlying TCP/TLS
connection to the Solace broker drops and JCSMP's internal background reconnect
(`TcpClientChannel$ClientChannelReconnect`) exhausts its retries, the worker
fails repeatedly with `com.google.cloud.RetryHelper$RetryHelperException:
com.solacesystems.jcsmp.StaleSessionException: Tried to start a closed message
consumer` and cannot recover the reader session.
#### Stack trace (anonymized)
```text
org.apache.beam.sdk.util.UserCodeException:
com.google.cloud.RetryHelper$RetryHelperException:
com.solacesystems.jcsmp.StaleSessionException: Tried to start a closed message
consumer.
at
org.apache.beam.sdk.util.UserCodeException.wrap(UserCodeException.java:39)
at
org.apache.beam.sdk.io.Read$UnboundedSourceAsSDFWrapperFn$DoFnInvoker.invokeProcessElement(Unknown
Source)
at
org.apache.beam.fn.harness.FnApiDoFnRunner.processElementForWindowObservingSizedElementAndRestriction(FnApiDoFnRunner.java:666)
at
org.apache.beam.fn.harness.FnApiDoFnRunner$1.accept(FnApiDoFnRunner.java:469)
at
org.apache.beam.fn.harness.FnApiDoFnRunner$1.accept(FnApiDoFnRunner.java:466)
at
org.apache.beam.fn.harness.data.PCollectionConsumerRegistry$MetricTrackingFnDataReceiver.accept(PCollectionConsumerRegistry.java:375)
at
org.apache.beam.fn.harness.data.PCollectionConsumerRegistry$MetricTrackingFnDataReceiver.accept(PCollectionConsumerRegistry.java:303)
at
org.apache.beam.fn.harness.BeamFnDataReadRunner.forwardElementToConsumer(BeamFnDataReadRunner.java:228)
at
org.apache.beam.sdk.fn.data.BeamFnDataInboundObserver.multiplexElements(BeamFnDataInboundObserver.java:233)
at
org.apache.beam.fn.harness.control.ProcessBundleHandler.processBundle(ProcessBundleHandler.java:540)
at
org.apache.beam.fn.harness.control.BeamFnControlClient.delegateOnInstructionRequestType(BeamFnControlClient.java:150)
at
org.apache.beam.fn.harness.control.BeamFnControlClient$InboundObserver.lambda$onNext$0(BeamFnControlClient.java:115)
at
java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:545)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:328)
at
org.apache.beam.sdk.util.UnboundedScheduledExecutorService$ScheduledFutureTask.run(UnboundedScheduledExecutorService.java:164)
at
java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1090)
at
java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:614)
at java.base/java.lang.Thread.run(Thread.java:1474)
Caused by: com.google.cloud.RetryHelper$RetryHelperException:
com.solacesystems.jcsmp.StaleSessionException: Tried to start a closed message
consumer.
at com.google.cloud.RetryHelper.runWithRetries(RetryHelper.java:54)
at
org.apache.beam.sdk.io.solace.RetryCallableManager.retryCallable(RetryCallableManager.java:66)
at
org.apache.beam.sdk.io.solace.broker.SolaceMessageReceiver.startFlowReceiver(SolaceMessageReceiver.java:43)
at
org.apache.beam.sdk.io.solace.broker.SolaceMessageReceiver.receive(SolaceMessageReceiver.java:56)
at
org.apache.beam.sdk.io.solace.read.UnboundedSolaceReader.advance(UnboundedSolaceReader.java:168)
at
org.apache.beam.sdk.io.Read$UnboundedSourceAsSDFWrapperFn$UnboundedSourceAsSDFRestrictionTracker.tryClaimOrThrow(Read.java:963)
at
org.apache.beam.sdk.io.Read$UnboundedSourceAsSDFWrapperFn$UnboundedSourceAsSDFRestrictionTracker.tryClaim(Read.java:933)
at
org.apache.beam.sdk.io.Read$UnboundedSourceAsSDFWrapperFn$UnboundedSourceAsSDFRestrictionTracker.tryClaim(Read.java:865)
at
org.apache.beam.sdk.fn.splittabledofn.RestrictionTrackers$RestrictionTrackerObserver.tryClaim(RestrictionTrackers.java:78)
at
org.apache.beam.sdk.io.Read$UnboundedSourceAsSDFWrapperFn.processElement(Read.java:593)
Caused by: com.solacesystems.jcsmp.StaleSessionException: Tried to start a
closed message consumer.
at
com.solacesystems.jcsmp.impl.flow.FlowHandleImpl.throwClosedException(FlowHandleImpl.java:2269)
at
com.solacesystems.jcsmp.impl.flow.FlowHandleImpl.startImpl(FlowHandleImpl.java:1149)
at
com.solacesystems.jcsmp.impl.flow.FlowHandleImpl.start(FlowHandleImpl.java:1134)
at
org.apache.beam.sdk.io.solace.broker.SolaceMessageReceiver.lambda$startFlowReceiver$0(SolaceMessageReceiver.java:45)
at
com.google.api.gax.retrying.DirectRetryingExecutor.submit(DirectRetryingExecutor.java:104)
at com.google.cloud.RetryHelper.run(RetryHelper.java:76)
at com.google.cloud.RetryHelper.runWithRetries(RetryHelper.java:50)
at
org.apache.beam.sdk.io.solace.RetryCallableManager.retryCallable(RetryCallableManager.java:66)
at
org.apache.beam.sdk.io.solace.broker.SolaceMessageReceiver.startFlowReceiver(SolaceMessageReceiver.java:43)
at
org.apache.beam.sdk.io.solace.broker.SolaceMessageReceiver.receive(SolaceMessageReceiver.java:56)
... 10 more
Caused by: com.solacesystems.jcsmp.JCSMPTransportException:
(JCSMPTransportException) Channel is closed by peer
at
com.solacesystems.jcsmp.protocol.impl.TcpClientChannel$ClientChannelReconnect.call(TcpClientChannel.java:2580)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:328)
... 3 more
Caused by: com.solacesystems.jcsmp.JCSMPTransportException: (Client name:
<client-name> local(/10.0.0.1:50000)
remote(solace-broker.example.com/192.0.2.1:55443)) - Error communicating with
the router.
at
com.solacesystems.jcsmp.protocol.impl.TcpChannel.sendLoginRequestWaitForLoginResponse(TcpChannel.java:254)
at
com.solacesystems.jcsmp.protocol.impl.ChannelOpStrategyClient.performOpen(ChannelOpStrategyClient.java:98)
at
com.solacesystems.jcsmp.protocol.impl.TcpClientChannel.performOpenSingle(TcpClientChannel.java:398)
at
com.solacesystems.jcsmp.protocol.impl.TcpClientChannel.access$1000(TcpClientChannel.java:133)
at
com.solacesystems.jcsmp.protocol.impl.TcpClientChannel$ClientChannelReconnect.call(TcpClientChannel.java:2425)
... 4 more
Caused by: java.nio.channels.ClosedChannelException
at io.netty.handler.ssl.SslHandler.channelInactive(SslHandler.java:1173)
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:303)
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
at
io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274)
at
io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1394)
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:301)
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
at
io.netty.channel.DefaultChannelPipeline.fireChannelInactive(DefaultChannelPipeline.java:892)
at
io.netty.channel.AbstractChannel$AbstractUnsafe$7.run(AbstractChannel.java:811)
at
io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
at
io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)
at
io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:405)
at
io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:1007)
at
io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at
io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
... 1 more
Suppressed: io.netty.handler.ssl.StacklessSSLHandshakeException:
Connection closed while SSL/TLS handshake was in progress
at io.netty.handler.ssl.SslHandler.channelInactive(Unknown
Source)
```
---
### Root Cause Analysis (3 Bugs)
#### 1. `SolaceMessageReceiver.receive()` tries to `.start()` a permanently
closed `FlowReceiver`
When JCSMP's `ClientChannelReconnect` exhausts its reconnect retries, it
aborts the session reconnect (`SubFlowManagerImpl.dispatchChannelException`)
and closes all active `FlowHandleImpl` instances (`opened = false`,
`lastException = reconnectException`).
When `flowReceiver.receive()` is subsequently called on the stale flow,
JCSMP throws `StaleSessionException("Tried to call receive on a closed message
consumer", lastException)`. `SolaceMessageReceiver.receive()` catches
`StaleSessionException` and calls `startFlowReceiver()`, which invokes
`flowReceiver.start()` on the **same** closed `FlowReceiver` instance:
```java
try {
return flowReceiver.receive(DEFAULT_ADVANCE_TIMEOUT_IN_MILLIS);
} catch (StaleSessionException e) {
startFlowReceiver();
throw new IOException(
"SolaceIO: Caught StaleSessionException, restarting the
FlowReceiver.", e);
}
```
In JCSMP, `FlowReceiver.start()` only resumes a stopped flow; a closed flow
(`opened == false`) is terminal and cannot be restarted. The first check in
`FlowHandleImpl.startImpl()` (`FlowHandleImpl.java:1149`) is `if (!this.opened)
throwClosedException("Tried to start a closed message consumer.")`, which
immediately throws `StaleSessionException` again. Because `startFlowReceiver()`
wraps `flowReceiver.start()` in `RetryCallableManager.retryCallable()`, it
retries `flowReceiver.start()` 4 times on the dead consumer (sleeping ~7s
total) and then throws an unchecked
`com.google.cloud.RetryHelper$RetryHelperException`.
#### 2. Unchecked `RetryHelperException` bypasses
`UnboundedSolaceReader.advance()` and leaks `ackExecutor` threads
`UnboundedSolaceReader.advance()` only catches `IOException`:
```java
try {
receivedXmlMessage = getSessionService().getReceiver().receive();
} catch (IOException e) {
LOG.warn("SolaceIO.Read: Exception when pulling messages from the
broker.", e);
return false;
}
```
When `RetryHelperException` (a `RuntimeException`) is thrown by
`startFlowReceiver()`, it escapes `advance()` and propagates into
`Read.UnboundedSourceAsSDFRestrictionTracker.tryClaim()`. Because
`Read.tryClaim()` only catches `IOException` to close `currentReader`, and
`initializeCurrentReader()` has already removed `currentReader` from
`cachedReaders` at the start of the bundle, `UnboundedSolaceReader.close()` is
**never called**. Every failed bundle leaks the reader's 4-thread `ackExecutor`
(`Executors.newFixedThreadPool(4)`).
#### 3. `JcsmpSessionService.getReceiver()` and `createFlowReceiver()`
cannot recover a stale receiver or aborted JCSMP session
Even if `SolaceMessageReceiver.receive()` threw `IOException` instead of
calling `startFlowReceiver()`, `JcsmpSessionService` cannot recover on
subsequent `advance()` calls:
```java
@Override
public synchronized MessageReceiver getReceiver() {
if (this.messageReceiver == null || this.jcsmpSession == null ||
this.jcsmpSession.isClosed()) {
this.messageReceiver =
retryCallableManager.retryCallable(
this::createFlowReceiver,
ImmutableSet.of(JCSMPException.class));
this.messageReceiver.start();
}
return checkStateNotNull(this.messageReceiver);
}
```
1. **Aborted JCSMP sessions do not report `isClosed() == true`:** In
`JCSMPBasicSession`, `isClosed()` only returns `true` when `closeSession()` was
explicitly called by the client. When background reconnect aborts due to
transport errors, JCSMP sets `isSessionAborted = true`
(`notifySessionReconnectAborted`), leaving `isClosed() == false`.
2. **`messageReceiver` is never checked for closure:** `getReceiver()` does
not check whether `this.messageReceiver` (or its underlying `FlowReceiver`) is
closed (`flowReceiver.isClosed()`), so `getReceiver()` keeps returning the same
dead `SolaceMessageReceiver` on every subsequent `advance()` call.
3. **`createFlowReceiver()` does not reconnect a closed/aborted session:**
Unlike `createXMLMessageProducer()` (which reconnects the session when
`jcsmpSession == null || jcsmpSession.isClosed()`), `createFlowReceiver()`
throws `IOException("SolaceIO.Read: Could not create a receiver from the Jcsmp
session: session object is null or closed.")` when the session is null or
closed, and `getReceiver()` only retries `JCSMPException.class`, not
`IOException.class`.
### Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
### Issue Components
- [x] Component: Java SDK
- [x] Component: IO connector
--
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]
