Skip to content

Commit c4f2daa

Browse files
vbabaninvbabanin
authored
Fix Netty event loop resource leak on MongoClient close. (#1646) (#1674)
JAVA-5812
1 parent 5fbb1ff commit c4f2daa

File tree

8 files changed

+201
-31
lines changed

8 files changed

+201
-31
lines changed

‎driver-core/src/main/com/mongodb/internal/connection/StreamFactoryHelper.java

Lines changed: 18 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,10 @@
2121
import com.mongodb.connection.AsyncTransportSettings;
2222
import com.mongodb.connection.NettyTransportSettings;
2323
import com.mongodb.connection.SocketSettings;
24+
import com.mongodb.connection.SslSettings;
2425
import com.mongodb.connection.TransportSettings;
2526
import com.mongodb.internal.connection.netty.NettyStreamFactoryFactory;
27+
import com.mongodb.lang.Nullable;
2628
import com.mongodb.spi.dns.InetAddressResolver;
2729

2830
import java.io.IOException;
@@ -34,16 +36,26 @@
3436
*/
3537
public final class StreamFactoryHelper {
3638

37-
public static StreamFactory getSyncStreamFactory(final MongoClientSettings settings,
38-
final InetAddressResolver inetAddressResolver, final SocketSettings socketSettings) {
39-
TransportSettings transportSettings = settings.getTransportSettings();
39+
public static StreamFactoryFactory getSyncStreamFactoryFactory(
40+
@Nullable final TransportSettings transportSettings,
41+
final InetAddressResolver inetAddressResolver) {
42+
4043
if (transportSettings == null) {
41-
return new SocketStreamFactory(inetAddressResolver, socketSettings, settings.getSslSettings());
44+
return new StreamFactoryFactory() {
45+
@Override
46+
public StreamFactory create(final SocketSettings socketSettings, final SslSettings sslSettings) {
47+
return new SocketStreamFactory(inetAddressResolver, socketSettings, sslSettings);
48+
}
49+
50+
@Override
51+
public void close() {
52+
//NOP
53+
}
54+
};
4255
} else if (transportSettings instanceof AsyncTransportSettings) {
4356
throw new MongoClientException("Unsupported transport settings in sync: " + transportSettings.getClass().getName());
4457
} else if (transportSettings instanceof NettyTransportSettings) {
45-
return getNettyStreamFactoryFactory(inetAddressResolver, (NettyTransportSettings) transportSettings)
46-
.create(socketSettings, settings.getSslSettings());
58+
return getNettyStreamFactoryFactory(inetAddressResolver, (NettyTransportSettings) transportSettings);
4759
} else {
4860
throw new MongoClientException("Unsupported transport settings: " + transportSettings.getClass().getName());
4961
}

‎driver-legacy/src/main/com/mongodb/MongoClient.java

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import com.mongodb.client.ListDatabasesIterable;
2222
import com.mongodb.client.MongoDatabase;
2323
import com.mongodb.client.MongoIterable;
24+
import com.mongodb.client.internal.Clusters;
2425
import com.mongodb.client.internal.MongoClientImpl;
2526
import com.mongodb.client.internal.OperationExecutor;
2627
import com.mongodb.connection.ClusterConnectionMode;
@@ -37,6 +38,7 @@
3738
import com.mongodb.internal.connection.Connection;
3839
import com.mongodb.internal.connection.NoOpSessionContext;
3940
import com.mongodb.internal.connection.OperationContext;
41+
import com.mongodb.internal.connection.StreamFactoryFactory;
4042
import com.mongodb.internal.diagnostics.logging.Logger;
4143
import com.mongodb.internal.diagnostics.logging.Loggers;
4244
import com.mongodb.internal.session.ServerSessionPool;
@@ -63,8 +65,11 @@
6365
import java.util.concurrent.atomic.AtomicBoolean;
6466
import java.util.stream.Collectors;
6567

68+
import static com.mongodb.assertions.Assertions.notNull;
6669
import static com.mongodb.internal.connection.ClientMetadataHelper.createClientMetadataDocument;
6770
import static com.mongodb.internal.connection.ServerAddressHelper.createServerAddress;
71+
import static com.mongodb.internal.connection.ServerAddressHelper.getInetAddressResolver;
72+
import static com.mongodb.internal.connection.StreamFactoryHelper.getSyncStreamFactoryFactory;
6873
import static java.lang.String.format;
6974
import static java.util.Collections.singletonList;
7075
import static java.util.concurrent.TimeUnit.SECONDS;
@@ -244,8 +249,20 @@ public MongoClient(final MongoClientSettings settings, @Nullable final MongoDriv
244249
private MongoClient(final MongoClientSettings settings,
245250
@Nullable final MongoClientOptions options,
246251
@Nullable final MongoDriverInformation mongoDriverInformation) {
252+
notNull("settings", settings);
253+
247254
MongoDriverInformation wrappedMongoDriverInformation = wrapMongoDriverInformation(mongoDriverInformation);
248-
delegate = new MongoClientImpl(settings, wrappedMongoDriverInformation);
255+
256+
StreamFactoryFactory syncStreamFactoryFactory = getSyncStreamFactoryFactory(
257+
settings.getTransportSettings(),
258+
getInetAddressResolver(settings));
259+
260+
Cluster cluster = Clusters.createCluster(
261+
settings,
262+
wrappedMongoDriverInformation,
263+
syncStreamFactoryFactory);
264+
265+
delegate = new MongoClientImpl(cluster, settings, wrappedMongoDriverInformation, syncStreamFactoryFactory);
249266
this.options = options != null ? options : MongoClientOptions.builder(settings).build();
250267
cursorCleaningService = this.options.isCursorFinalizerEnabled() ? createCursorCleaningService() : null;
251268
this.closed = new AtomicBoolean();

‎driver-legacy/src/test/unit/com/mongodb/MongoClientSpecification.groovy

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -309,7 +309,7 @@ class MongoClientSpecification extends Specification {
309309
def 'should validate the ChangeStreamIterable pipeline data correctly'() {
310310
given:
311311
def executor = new TestOperationExecutor([])
312-
def client = new MongoClientImpl(Stub(Cluster), null, MongoClientSettings.builder().build(), executor)
312+
def client = new MongoClientImpl(Stub(Cluster), null, MongoClientSettings.builder().build(), null, executor)
313313

314314
when:
315315
client.watch((Class) null)

‎driver-sync/src/main/com/mongodb/client/MongoClients.java

Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,9 +19,16 @@
1919
import com.mongodb.ConnectionString;
2020
import com.mongodb.MongoClientSettings;
2121
import com.mongodb.MongoDriverInformation;
22+
import com.mongodb.client.internal.Clusters;
2223
import com.mongodb.client.internal.MongoClientImpl;
24+
import com.mongodb.internal.connection.Cluster;
25+
import com.mongodb.internal.connection.StreamFactoryFactory;
2326
import com.mongodb.lang.Nullable;
2427

28+
import static com.mongodb.assertions.Assertions.notNull;
29+
import static com.mongodb.internal.connection.ServerAddressHelper.getInetAddressResolver;
30+
import static com.mongodb.internal.connection.StreamFactoryHelper.getSyncStreamFactoryFactory;
31+
2532

2633
/**
2734
* A factory for {@link MongoClient} instances. Use of this class is now the recommended way to connect to MongoDB via the Java driver.
@@ -103,9 +110,23 @@ public static MongoClient create(final ConnectionString connectionString,
103110
* @return the client
104111
*/
105112
public static MongoClient create(final MongoClientSettings settings, @Nullable final MongoDriverInformation mongoDriverInformation) {
113+
notNull("settings", settings);
114+
106115
MongoDriverInformation.Builder builder = mongoDriverInformation == null ? MongoDriverInformation.builder()
107116
: MongoDriverInformation.builder(mongoDriverInformation);
108-
return new MongoClientImpl(settings, builder.driverName("sync").build());
117+
118+
MongoDriverInformation driverInfo = builder.driverName("sync").build();
119+
120+
StreamFactoryFactory syncStreamFactoryFactory = getSyncStreamFactoryFactory(
121+
settings.getTransportSettings(),
122+
getInetAddressResolver(settings));
123+
124+
Cluster cluster = Clusters.createCluster(
125+
settings,
126+
driverInfo,
127+
syncStreamFactoryFactory);
128+
129+
return new MongoClientImpl(cluster, settings, driverInfo, syncStreamFactoryFactory);
109130
}
110131

111132
private MongoClients() {

‎driver-sync/src/main/com/mongodb/client/internal/Clusters.java

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
1+
/*
2+
* Copyright 2008-present MongoDB, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.mongodb.client.internal;
17+
18+
import com.mongodb.MongoClientSettings;
19+
import com.mongodb.MongoDriverInformation;
20+
import com.mongodb.connection.SocketSettings;
21+
import com.mongodb.internal.TimeoutSettings;
22+
import com.mongodb.internal.connection.Cluster;
23+
import com.mongodb.internal.connection.DefaultClusterFactory;
24+
import com.mongodb.internal.connection.InternalConnectionPoolSettings;
25+
import com.mongodb.internal.connection.StreamFactory;
26+
import com.mongodb.internal.connection.StreamFactoryFactory;
27+
import com.mongodb.lang.Nullable;
28+
29+
import static com.mongodb.assertions.Assertions.assertNotNull;
30+
import static com.mongodb.internal.event.EventListenerHelper.getCommandListener;
31+
32+
public final class Clusters {
33+
34+
private Clusters() {
35+
//NOP
36+
}
37+
38+
public static Cluster createCluster(final MongoClientSettings settings,
39+
@Nullable final MongoDriverInformation mongoDriverInformation,
40+
final StreamFactoryFactory streamFactoryFactory) {
41+
assertNotNull(streamFactoryFactory);
42+
assertNotNull(settings);
43+
44+
StreamFactory streamFactory = getStreamFactory(streamFactoryFactory, settings, false);
45+
StreamFactory heartbeatStreamFactory = getStreamFactory(streamFactoryFactory, settings, true);
46+
47+
return new DefaultClusterFactory().createCluster(settings.getClusterSettings(), settings.getServerSettings(),
48+
settings.getConnectionPoolSettings(), InternalConnectionPoolSettings.builder().build(),
49+
TimeoutSettings.create(settings), streamFactory,
50+
TimeoutSettings.createHeartbeatSettings(settings), heartbeatStreamFactory,
51+
settings.getCredential(), settings.getLoggerSettings(), getCommandListener(settings.getCommandListeners()),
52+
settings.getApplicationName(), mongoDriverInformation, settings.getCompressorList(), settings.getServerApi(),
53+
settings.getDnsClient());
54+
}
55+
56+
private static StreamFactory getStreamFactory(
57+
final StreamFactoryFactory streamFactoryFactory,
58+
final MongoClientSettings settings,
59+
final boolean isHeartbeat) {
60+
SocketSettings socketSettings = isHeartbeat ? settings.getHeartbeatSocketSettings() : settings.getSocketSettings();
61+
return streamFactoryFactory.create(socketSettings, settings.getSslSettings());
62+
}
63+
}

‎driver-sync/src/main/com/mongodb/client/internal/MongoClientImpl.java

Lines changed: 30 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -33,20 +33,20 @@
3333
import com.mongodb.client.MongoIterable;
3434
import com.mongodb.client.SynchronousContextProvider;
3535
import com.mongodb.client.model.bulk.ClientBulkWriteOptions;
36-
import com.mongodb.client.model.bulk.ClientNamespacedWriteModel;
3736
import com.mongodb.client.model.bulk.ClientBulkWriteResult;
37+
import com.mongodb.client.model.bulk.ClientNamespacedWriteModel;
3838
import com.mongodb.connection.ClusterDescription;
3939
import com.mongodb.connection.SocketSettings;
4040
import com.mongodb.internal.TimeoutSettings;
4141
import com.mongodb.internal.connection.Cluster;
4242
import com.mongodb.internal.connection.DefaultClusterFactory;
4343
import com.mongodb.internal.connection.InternalConnectionPoolSettings;
4444
import com.mongodb.internal.connection.StreamFactory;
45+
import com.mongodb.internal.connection.StreamFactoryFactory;
4546
import com.mongodb.internal.diagnostics.logging.Logger;
4647
import com.mongodb.internal.diagnostics.logging.Loggers;
4748
import com.mongodb.internal.session.ServerSessionPool;
4849
import com.mongodb.lang.Nullable;
49-
import com.mongodb.spi.dns.InetAddressResolver;
5050
import org.bson.BsonDocument;
5151
import org.bson.Document;
5252
import org.bson.codecs.configuration.CodecRegistry;
@@ -59,8 +59,6 @@
5959
import static com.mongodb.assertions.Assertions.notNull;
6060
import static com.mongodb.client.internal.Crypts.createCrypt;
6161
import static com.mongodb.internal.connection.ClientMetadataHelper.createClientMetadataDocument;
62-
import static com.mongodb.internal.connection.ServerAddressHelper.getInetAddressResolver;
63-
import static com.mongodb.internal.connection.StreamFactoryHelper.getSyncStreamFactory;
6462
import static com.mongodb.internal.event.EventListenerHelper.getCommandListener;
6563
import static java.lang.String.format;
6664
import static org.bson.codecs.configuration.CodecRegistries.withUuidRepresentation;
@@ -75,14 +73,22 @@ public final class MongoClientImpl implements MongoClient {
7573
private final MongoDriverInformation mongoDriverInformation;
7674
private final MongoClusterImpl delegate;
7775
private final AtomicBoolean closed;
76+
private final AutoCloseable externalResourceCloser;
7877

79-
public MongoClientImpl(final MongoClientSettings settings, final MongoDriverInformation mongoDriverInformation) {
80-
this(createCluster(settings, mongoDriverInformation), mongoDriverInformation, settings, null);
78+
public MongoClientImpl(final Cluster cluster,
79+
final MongoClientSettings settings,
80+
final MongoDriverInformation mongoDriverInformation,
81+
@Nullable final AutoCloseable externalResourceCloser) {
82+
this(cluster, mongoDriverInformation, settings, externalResourceCloser, null);
8183
}
8284

83-
public MongoClientImpl(final Cluster cluster, final MongoDriverInformation mongoDriverInformation,
85+
private MongoClientImpl(final Cluster cluster,
86+
final MongoDriverInformation mongoDriverInformation,
8487
final MongoClientSettings settings,
85-
@Nullable final OperationExecutor operationExecutor) {
88+
@Nullable final AutoCloseable externalResourceCloser,
89+
@Nullable final OperationExecutor operationExecutor) {
90+
91+
this.externalResourceCloser = externalResourceCloser;
8692
this.settings = notNull("settings", settings);
8793
this.mongoDriverInformation = mongoDriverInformation;
8894
AutoEncryptionSettings autoEncryptionSettings = settings.getAutoEncryptionSettings();
@@ -114,6 +120,13 @@ public void close() {
114120
}
115121
delegate.getServerSessionPool().close();
116122
delegate.getCluster().close();
123+
if (externalResourceCloser != null) {
124+
try {
125+
externalResourceCloser.close();
126+
} catch (Exception e) {
127+
LOGGER.warn("Exception closing resource", e);
128+
}
129+
}
117130
}
118131
}
119132

@@ -287,21 +300,24 @@ public ClientBulkWriteResult bulkWrite(
287300
}
288301

289302
private static Cluster createCluster(final MongoClientSettings settings,
290-
@Nullable final MongoDriverInformation mongoDriverInformation) {
303+
@Nullable final MongoDriverInformation mongoDriverInformation,
304+
final StreamFactory streamFactory, final StreamFactory heartbeatStreamFactory) {
291305
notNull("settings", settings);
292306
return new DefaultClusterFactory().createCluster(settings.getClusterSettings(), settings.getServerSettings(),
293307
settings.getConnectionPoolSettings(), InternalConnectionPoolSettings.builder().build(),
294-
TimeoutSettings.create(settings), getStreamFactory(settings, false),
295-
TimeoutSettings.createHeartbeatSettings(settings), getStreamFactory(settings, true),
308+
TimeoutSettings.create(settings), streamFactory,
309+
TimeoutSettings.createHeartbeatSettings(settings), heartbeatStreamFactory,
296310
settings.getCredential(), settings.getLoggerSettings(), getCommandListener(settings.getCommandListeners()),
297311
settings.getApplicationName(), mongoDriverInformation, settings.getCompressorList(), settings.getServerApi(),
298312
settings.getDnsClient());
299313
}
300314

301-
private static StreamFactory getStreamFactory(final MongoClientSettings settings, final boolean isHeartbeat) {
315+
private static StreamFactory getStreamFactory(
316+
final StreamFactoryFactory streamFactoryFactory,
317+
final MongoClientSettings settings,
318+
final boolean isHeartbeat) {
302319
SocketSettings socketSettings = isHeartbeat ? settings.getHeartbeatSocketSettings() : settings.getSocketSettings();
303-
InetAddressResolver inetAddressResolver = getInetAddressResolver(settings);
304-
return getSyncStreamFactory(settings, inetAddressResolver, socketSettings);
320+
return streamFactoryFactory.create(socketSettings, settings.getSslSettings());
305321
}
306322

307323
public Cluster getCluster() {

‎driver-sync/src/test/functional/com/mongodb/client/MongoClientTest.java

Lines changed: 42 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,10 +17,16 @@
1717
package com.mongodb.client;
1818

1919
import com.mongodb.ClusterFixture;
20+
import com.mongodb.MongoClientSettings;
21+
import com.mongodb.MongoDriverInformation;
22+
import com.mongodb.client.internal.MongoClientImpl;
2023
import com.mongodb.connection.ClusterId;
2124
import com.mongodb.event.ClusterListener;
2225
import com.mongodb.event.ClusterOpeningEvent;
26+
import com.mongodb.internal.connection.Cluster;
27+
import com.mongodb.internal.mockito.MongoMockito;
2328
import org.junit.jupiter.api.Test;
29+
import org.mockito.Mockito;
2430

2531
import java.util.concurrent.CompletableFuture;
2632
import java.util.concurrent.ExecutionException;
@@ -29,12 +35,13 @@
2935

3036
import static com.mongodb.client.Fixture.getMongoClientSettingsBuilder;
3137
import static org.junit.jupiter.api.Assertions.assertEquals;
38+
import static org.mockito.Mockito.doNothing;
3239

33-
public class MongoClientTest {
40+
class MongoClientTest {
3441

3542
@SuppressWarnings("try")
3643
@Test
37-
public void shouldIncludeApplicationNameInClusterId() throws InterruptedException,
44+
void shouldIncludeApplicationNameInClusterId() throws InterruptedException,
3845
ExecutionException, TimeoutException {
3946
CompletableFuture<ClusterId> clusterIdFuture = new CompletableFuture<>();
4047
ClusterListener clusterListener = new ClusterListener() {
@@ -52,4 +59,37 @@ public void clusterOpening(final ClusterOpeningEvent event) {
5259
assertEquals(applicationName, clusterId.getDescription());
5360
}
5461
}
62+
63+
@Test
64+
void shouldCloseExternalResources() throws Exception {
65+
66+
//given
67+
Cluster cluster = MongoMockito.mock(
68+
Cluster.class,
69+
mockedCluster -> {
70+
doNothing().when(mockedCluster).close();
71+
});
72+
AutoCloseable externalResource = MongoMockito.mock(
73+
AutoCloseable.class,
74+
mockedExternalResource -> {
75+
try {
76+
doNothing().when(mockedExternalResource).close();
77+
} catch (Exception e) {
78+
throw new RuntimeException(e);
79+
}
80+
});
81+
82+
MongoClientImpl mongoClient = new MongoClientImpl(
83+
cluster,
84+
MongoClientSettings.builder().build(),
85+
MongoDriverInformation.builder().build(),
86+
externalResource);
87+
88+
//when
89+
mongoClient.close();
90+
91+
//then
92+
Mockito.verify(externalResource).close();
93+
Mockito.verify(cluster).close();
94+
}
5595
}

0 commit comments

Comments
 (0)