Skip to content

Commit feabdf7

Browse files
adutraadutra
committed
JAVA-2943: Prevent session leak with wrong keyspace name
1 parent 4d29623 commit feabdf7

3 files changed

Lines changed: 72 additions & 38 deletions

File tree

‎changelog/README.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44

55
### 4.11.2 (in progress)
66

7+
- [bug] JAVA-2943: Prevent session leak with wrong keyspace name
78
- [bug] JAVA-2938: OverloadedException message is misleading
89

910
### 4.11.1

‎core/src/main/java/com/datastax/oss/driver/internal/core/session/DefaultSession.java‎

Lines changed: 43 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -384,16 +384,30 @@ private void init(CqlIdentifier keyspace) {
384384
.getTopologyMonitor()
385385
.init()
386386
.thenCompose(v -> metadataManager.refreshNodes())
387-
.thenAccept(v -> afterInitialNodeListRefresh(keyspace))
388-
.exceptionally(
389-
error -> {
390-
initFuture.completeExceptionally(error);
391-
RunOrSchedule.on(adminExecutor, this::close);
392-
return null;
387+
.thenCompose(v -> checkProtocolVersion())
388+
.thenCompose(v -> initialSchemaRefresh())
389+
.thenCompose(v -> initializePools(keyspace))
390+
.whenComplete(
391+
(v, error) -> {
392+
if (error == null) {
393+
LOG.debug("[{}] Initialization complete, ready", logPrefix);
394+
notifyListeners();
395+
initFuture.complete(DefaultSession.this);
396+
} else {
397+
LOG.debug("[{}] Initialization failed, force closing", logPrefix, error);
398+
forceCloseAsync()
399+
.whenComplete(
400+
(v1, error1) -> {
401+
if (error1 != null) {
402+
error.addSuppressed(error1);
403+
}
404+
initFuture.completeExceptionally(error);
405+
});
406+
}
393407
});
394408
}
395409

396-
private void afterInitialNodeListRefresh(CqlIdentifier keyspace) {
410+
private CompletionStage<Void> checkProtocolVersion() {
397411
try {
398412
boolean protocolWasForced =
399413
context.getConfig().getDefaultProfile().isDefined(DefaultDriverOption.PROTOCOL_VERSION);
@@ -426,48 +440,39 @@ private void afterInitialNodeListRefresh(CqlIdentifier keyspace) {
426440
bestVersion);
427441
}
428442
}
429-
metadataManager
443+
return CompletableFuture.completedFuture(null);
444+
} catch (Throwable throwable) {
445+
return CompletableFutures.failedFuture(throwable);
446+
}
447+
}
448+
449+
private CompletionStage<RefreshSchemaResult> initialSchemaRefresh() {
450+
try {
451+
return metadataManager
430452
.refreshSchema(null, false, true)
431-
.whenComplete(
432-
(metadata, error) -> {
433-
if (error != null) {
434-
Loggers.warnWithException(
435-
LOG,
436-
"[{}] Unexpected error while refreshing schema during initialization, "
437-
+ "keeping previous version",
438-
logPrefix,
439-
error);
440-
}
441-
afterInitialSchemaRefresh(keyspace);
453+
.exceptionally(
454+
error -> {
455+
Loggers.warnWithException(
456+
LOG,
457+
"[{}] Unexpected error while refreshing schema during initialization, "
458+
+ "proceeding without schema metadata",
459+
logPrefix,
460+
error);
461+
return null;
442462
});
443463
} catch (Throwable throwable) {
444-
initFuture.completeExceptionally(throwable);
464+
return CompletableFutures.failedFuture(throwable);
445465
}
446466
}
447467

448-
private void afterInitialSchemaRefresh(CqlIdentifier keyspace) {
468+
private CompletionStage<Void> initializePools(CqlIdentifier keyspace) {
449469
try {
450470
nodeStateManager.markInitialized();
451471
context.getLoadBalancingPolicyWrapper().init();
452472
context.getConfigLoader().onDriverInit(context);
453-
LOG.debug("[{}] Initialization complete, ready", logPrefix);
454-
poolManager
455-
.init(keyspace)
456-
.whenComplete(
457-
(v, error) -> {
458-
if (error != null) {
459-
initFuture.completeExceptionally(error);
460-
} else {
461-
notifyListeners();
462-
initFuture.complete(DefaultSession.this);
463-
}
464-
});
473+
return poolManager.init(keyspace);
465474
} catch (Throwable throwable) {
466-
forceCloseAsync()
467-
.whenComplete(
468-
(v, error) -> {
469-
initFuture.completeExceptionally(throwable);
470-
});
475+
return CompletableFutures.failedFuture(throwable);
471476
}
472477
}
473478

‎integration-tests/src/test/java/com/datastax/oss/driver/core/SessionLeakIT.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
package com.datastax.oss.driver.core;
1717

1818
import static org.assertj.core.api.Assertions.assertThat;
19+
import static org.assertj.core.api.Fail.fail;
1920
import static org.mockito.ArgumentMatchers.any;
2021
import static org.mockito.Mockito.never;
2122
import static org.mockito.Mockito.reset;
@@ -25,7 +26,9 @@
2526
import ch.qos.logback.classic.Logger;
2627
import ch.qos.logback.classic.spi.ILoggingEvent;
2728
import ch.qos.logback.core.Appender;
29+
import com.datastax.oss.driver.api.core.CqlIdentifier;
2830
import com.datastax.oss.driver.api.core.CqlSession;
31+
import com.datastax.oss.driver.api.core.InvalidKeyspaceException;
2932
import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
3033
import com.datastax.oss.driver.api.core.config.DriverConfigLoader;
3134
import com.datastax.oss.driver.api.core.session.Session;
@@ -34,6 +37,7 @@
3437
import com.datastax.oss.driver.categories.IsolatedTests;
3538
import com.datastax.oss.driver.internal.core.session.DefaultSession;
3639
import com.datastax.oss.simulacron.common.cluster.ClusterSpec;
40+
import com.datastax.oss.simulacron.common.stubbing.PrimeDsl;
3741
import java.util.HashSet;
3842
import java.util.Set;
3943
import org.junit.Before;
@@ -103,4 +107,28 @@ public void should_warn_when_session_count_exceeds_threshold() {
103107
verify(appender, never()).doAppend(any());
104108
session.close();
105109
}
110+
111+
@Test
112+
public void should_never_warn_when_session_init_fails() {
113+
SIMULACRON_RULE
114+
.cluster()
115+
.prime(PrimeDsl.when("USE \"non_existent_keyspace\"").then(PrimeDsl.invalid("irrelevant")));
116+
int threshold = 4;
117+
// Set the config option explicitly, in case it gets overridden in the test application.conf:
118+
DriverConfigLoader configLoader =
119+
DriverConfigLoader.programmaticBuilder()
120+
.withInt(DefaultDriverOption.SESSION_LEAK_THRESHOLD, threshold)
121+
.build();
122+
// Go over the threshold, no warnings expected
123+
for (int i = 0; i < threshold + 1; i++) {
124+
try (Session session =
125+
SessionUtils.newSession(
126+
SIMULACRON_RULE, CqlIdentifier.fromCql("non_existent_keyspace"), configLoader)) {
127+
fail("Session %s should have failed to initialize", session.getName());
128+
} catch (InvalidKeyspaceException e) {
129+
assertThat(e.getMessage()).isEqualTo("Invalid keyspace non_existent_keyspace");
130+
}
131+
}
132+
verify(appender, never()).doAppend(any());
133+
}
106134
}

0 commit comments

Comments
 (0)