Repository navigation
[Bug]: Transient connection error on worker fails all subsequent work items in CassandraIO read #39780
Description
What happened?
In CassandraIO, when a worker fails to connect to Cassandra database while processing a work item, it throws an exception (like NoHostAvailableException) [1] and the work item fails. However, it was observed that all the subsequent work items ran on the same worker fail immediately with an IllegalStateException [2] without even attempting to connect to the database.
If a failed work item attempts all its leases on same worker, the entire job is failing due to a transient connectivity issue.
[1] - Initial Exception
at com.datastax.driver.core.ControlConnection.reconnectInternal(ControlConnection.java:270)
at com.datastax.driver.core.ControlConnection.connect(ControlConnection.java:109)
at com.datastax.driver.core.Cluster$Manager.negotiateProtocolVersionAndConnect(Cluster.java:1813)
at com.datastax.driver.core.Cluster$Manager.init(Cluster.java:1726)
at com.datastax.driver.core.Cluster.init(Cluster.java:214)
at com.datastax.driver.core.Cluster.connectAsync(Cluster.java:387)
at com.datastax.driver.core.Cluster.connect(Cluster.java:338)
at org.apache.beam.sdk.io.cassandra.ConnectionManager.lambda$getSession$3(ConnectionManager.java:78)
at java.base/java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1705)
at org.apache.beam.sdk.io.cassandra.ConnectionManager.getSession(ConnectionManager.java:76)```
[2] - Subsequent Exceptions (on same worker)
```java.lang.IllegalStateException: Can't use this cluster instance because it encountered an error in its initialization
at com.datastax.driver.core.Cluster.checkNotClosed(Cluster.java:650)
at com.datastax.driver.core.Cluster.connectAsync(Cluster.java:386)
at com.datastax.driver.core.Cluster.connect(Cluster.java:338)
at org.apache.beam.sdk.io.cassandra.ConnectionManager.lambda$getSession$3(ConnectionManager.java:78)
at java.base/java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1705)
at org.apache.beam.sdk.io.cassandra.ConnectionManager.getSession(ConnectionManager.java:76)```
### Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
### Issue Components
- [ ] Component: Python SDK
- [ ] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Prism Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner
Activity
sharan-malyala commentedon Aug 17, 2026
.take-issue
sharan-malyala commentedon Aug 17, 2026
Triage :
Ref - https://github.com/apache/beam/blob/master/sdks/java/io/cassandra/src/main/java/org/apache/beam/sdk/io/cassandra/ConnectionManager.java
getSessionfails on an attempt (due to any transient issue on Cassandra), the broken Cluster object remains in the cache.checkNotClosed()and immediately fails it with anIllegalStateExceptionbecause of the previous initialization failure.Proposed solution :
I will send a PR soon with the proposed change.
sharan-malyala commentedon Aug 17, 2026
cc @stankiewicz @damccorm