Abacn commented on code in PR #39803:
URL: https://github.com/apache/beam/pull/39803#discussion_r3806783396
##########
sdks/java/io/cassandra/src/main/java/org/apache/beam/sdk/io/cassandra/ConnectionManager.java:
##########
@@ -59,9 +59,22 @@ private static String readToSessionHash(Read<?> read) {
}
static Session getSession(Read<?> read) {
+ String clusterHash = readToClusterHash(read);
+ String sessionHash = readToSessionHash(read);
+
+ Cluster cachedCluster = clusterMap.get(clusterHash);
+
+ if (cachedCluster != null && cachedCluster.isClosed()) {
Review Comment:
I was thinking about the following scenario:
1. A session corrupted and cluster closed
2. Caller A runs getSession(), found cachedCluster.isClosed
3. Caller B also runs getSession(), found cachedCluster.isClosed
4. Caller A cleaned up session and cluster cache, recreated cluster
connection
5. Caller B clean up session again, because it's already inside the if
clause. But it should not do so as the cache is current valid, may cause
session leak.
Basically it seems the whole getSession now needs to be synchronized.
--
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]
