Skip to content

Commit 1b8f73f

Browse files
Tim-Brookskcm
Tim-Brooks
authored and
kcm
committed
Reduce channels in AbstractSimpleTransportTestCase (#34863)
This is related to #30876. The AbstractSimpleTransportTestCase initiates many tcp connections. There are normally over 1,000 connections in TIME_WAIT at the end of the test. This is because every test opens at least two different transports that connect to each other with 13 channel connection profiles. This commit modifies the default connection profile used by this test to 6. One connection for each type, except for REG which gets 2 connections.
1 parent f2664d7 commit 1b8f73f

File tree

11 files changed

+59
-38
lines changed

11 files changed

+59
-38
lines changed

‎modules/transport-netty4/src/test/java/org/elasticsearch/transport/netty4/SimpleNetty4TransportTests.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,7 @@ protected Version getCurrentVersion() {
7373
}
7474
};
7575
MockTransportService mockTransportService =
76-
MockTransportService.createNewService(Settings.EMPTY, transport, version, threadPool, clusterSettings, Collections.emptySet());
76+
MockTransportService.createNewService(settings, transport, version, threadPool, clusterSettings, Collections.emptySet());
7777
mockTransportService.start();
7878
return mockTransportService;
7979
}

‎plugins/transport-nio/src/test/java/org/elasticsearch/transport/nio/SimpleNioTransportTests.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -77,7 +77,7 @@ protected Version getCurrentVersion() {
7777
}
7878
};
7979
MockTransportService mockTransportService =
80-
MockTransportService.createNewService(Settings.EMPTY, transport, version, threadPool, clusterSettings, Collections.emptySet());
80+
MockTransportService.createNewService(settings, transport, version, threadPool, clusterSettings, Collections.emptySet());
8181
mockTransportService.start();
8282
return mockTransportService;
8383
}

‎server/src/main/java/org/elasticsearch/transport/ConnectionManager.java‎

Lines changed: 1 addition & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,7 @@ public class ConnectionManager implements Closeable {
6767
private final DelegatingNodeConnectionListener connectionListener = new DelegatingNodeConnectionListener();
6868

6969
public ConnectionManager(Settings settings, Transport transport, ThreadPool threadPool) {
70-
this(settings, transport, threadPool, buildDefaultConnectionProfile(settings));
70+
this(settings, transport, threadPool, ConnectionProfile.buildDefaultConnectionProfile(settings));
7171
}
7272

7373
public ConnectionManager(Settings settings, Transport transport, ThreadPool threadPool, ConnectionProfile defaultProfile) {
@@ -323,23 +323,4 @@ public void onConnectionClosed(Transport.Connection connection) {
323323
}
324324
}
325325
}
326-
327-
public static ConnectionProfile buildDefaultConnectionProfile(Settings settings) {
328-
int connectionsPerNodeRecovery = TransportService.CONNECTIONS_PER_NODE_RECOVERY.get(settings);
329-
int connectionsPerNodeBulk = TransportService.CONNECTIONS_PER_NODE_BULK.get(settings);
330-
int connectionsPerNodeReg = TransportService.CONNECTIONS_PER_NODE_REG.get(settings);
331-
int connectionsPerNodeState = TransportService.CONNECTIONS_PER_NODE_STATE.get(settings);
332-
int connectionsPerNodePing = TransportService.CONNECTIONS_PER_NODE_PING.get(settings);
333-
ConnectionProfile.Builder builder = new ConnectionProfile.Builder();
334-
builder.setConnectTimeout(TransportService.TCP_CONNECT_TIMEOUT.get(settings));
335-
builder.setHandshakeTimeout(TransportService.TCP_CONNECT_TIMEOUT.get(settings));
336-
builder.addConnections(connectionsPerNodeBulk, TransportRequestOptions.Type.BULK);
337-
builder.addConnections(connectionsPerNodePing, TransportRequestOptions.Type.PING);
338-
// if we are not master eligible we don't need a dedicated channel to publish the state
339-
builder.addConnections(DiscoveryNode.isMasterNode(settings) ? connectionsPerNodeState : 0, TransportRequestOptions.Type.STATE);
340-
// if we are not a data-node we don't need any dedicated channels for recovery
341-
builder.addConnections(DiscoveryNode.isDataNode(settings) ? connectionsPerNodeRecovery : 0, TransportRequestOptions.Type.RECOVERY);
342-
builder.addConnections(connectionsPerNodeReg, TransportRequestOptions.Type.REG);
343-
return builder.build();
344-
}
345326
}

‎server/src/main/java/org/elasticsearch/transport/ConnectionProfile.java‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,9 @@
1818
*/
1919
package org.elasticsearch.transport;
2020

21+
import org.elasticsearch.cluster.node.DiscoveryNode;
2122
import org.elasticsearch.common.Nullable;
23+
import org.elasticsearch.common.settings.Settings;
2224
import org.elasticsearch.common.unit.TimeValue;
2325

2426
import java.util.ArrayList;
@@ -91,6 +93,31 @@ public static ConnectionProfile resolveConnectionProfile(@Nullable ConnectionPro
9193
}
9294
}
9395

96+
/**
97+
* Builds a default connection profile based on the provided settings.
98+
*
99+
* @param settings to build the connection profile from
100+
* @return the connection profile
101+
*/
102+
public static ConnectionProfile buildDefaultConnectionProfile(Settings settings) {
103+
int connectionsPerNodeRecovery = TransportService.CONNECTIONS_PER_NODE_RECOVERY.get(settings);
104+
int connectionsPerNodeBulk = TransportService.CONNECTIONS_PER_NODE_BULK.get(settings);
105+
int connectionsPerNodeReg = TransportService.CONNECTIONS_PER_NODE_REG.get(settings);
106+
int connectionsPerNodeState = TransportService.CONNECTIONS_PER_NODE_STATE.get(settings);
107+
int connectionsPerNodePing = TransportService.CONNECTIONS_PER_NODE_PING.get(settings);
108+
Builder builder = new Builder();
109+
builder.setConnectTimeout(TransportService.TCP_CONNECT_TIMEOUT.get(settings));
110+
builder.setHandshakeTimeout(TransportService.TCP_CONNECT_TIMEOUT.get(settings));
111+
builder.addConnections(connectionsPerNodeBulk, TransportRequestOptions.Type.BULK);
112+
builder.addConnections(connectionsPerNodePing, TransportRequestOptions.Type.PING);
113+
// if we are not master eligible we don't need a dedicated channel to publish the state
114+
builder.addConnections(DiscoveryNode.isMasterNode(settings) ? connectionsPerNodeState : 0, TransportRequestOptions.Type.STATE);
115+
// if we are not a data-node we don't need any dedicated channels for recovery
116+
builder.addConnections(DiscoveryNode.isDataNode(settings) ? connectionsPerNodeRecovery : 0, TransportRequestOptions.Type.RECOVERY);
117+
builder.addConnections(connectionsPerNodeReg, TransportRequestOptions.Type.REG);
118+
return builder.build();
119+
}
120+
94121
/**
95122
* A builder to build a new {@link ConnectionProfile}
96123
*/

‎server/src/test/java/org/elasticsearch/transport/ConnectionManagerTests.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ public void stopThreadPool() {
6464
}
6565

6666
public void testConnectionProfileResolve() {
67-
final ConnectionProfile defaultProfile = ConnectionManager.buildDefaultConnectionProfile(Settings.EMPTY);
67+
final ConnectionProfile defaultProfile = ConnectionProfile.buildDefaultConnectionProfile(Settings.EMPTY);
6868
assertEquals(defaultProfile, ConnectionProfile.resolveConnectionProfile(null, defaultProfile));
6969

7070
final ConnectionProfile.Builder builder = new ConnectionProfile.Builder();
@@ -96,31 +96,31 @@ public void testConnectionProfileResolve() {
9696
}
9797

9898
public void testDefaultConnectionProfile() {
99-
ConnectionProfile profile = ConnectionManager.buildDefaultConnectionProfile(Settings.EMPTY);
99+
ConnectionProfile profile = ConnectionProfile.buildDefaultConnectionProfile(Settings.EMPTY);
100100
assertEquals(13, profile.getNumConnections());
101101
assertEquals(1, profile.getNumConnectionsPerType(TransportRequestOptions.Type.PING));
102102
assertEquals(6, profile.getNumConnectionsPerType(TransportRequestOptions.Type.REG));
103103
assertEquals(1, profile.getNumConnectionsPerType(TransportRequestOptions.Type.STATE));
104104
assertEquals(2, profile.getNumConnectionsPerType(TransportRequestOptions.Type.RECOVERY));
105105
assertEquals(3, profile.getNumConnectionsPerType(TransportRequestOptions.Type.BULK));
106106

107-
profile = ConnectionManager.buildDefaultConnectionProfile(Settings.builder().put("node.master", false).build());
107+
profile = ConnectionProfile.buildDefaultConnectionProfile(Settings.builder().put("node.master", false).build());
108108
assertEquals(12, profile.getNumConnections());
109109
assertEquals(1, profile.getNumConnectionsPerType(TransportRequestOptions.Type.PING));
110110
assertEquals(6, profile.getNumConnectionsPerType(TransportRequestOptions.Type.REG));
111111
assertEquals(0, profile.getNumConnectionsPerType(TransportRequestOptions.Type.STATE));
112112
assertEquals(2, profile.getNumConnectionsPerType(TransportRequestOptions.Type.RECOVERY));
113113
assertEquals(3, profile.getNumConnectionsPerType(TransportRequestOptions.Type.BULK));
114114

115-
profile = ConnectionManager.buildDefaultConnectionProfile(Settings.builder().put("node.data", false).build());
115+
profile = ConnectionProfile.buildDefaultConnectionProfile(Settings.builder().put("node.data", false).build());
116116
assertEquals(11, profile.getNumConnections());
117117
assertEquals(1, profile.getNumConnectionsPerType(TransportRequestOptions.Type.PING));
118118
assertEquals(6, profile.getNumConnectionsPerType(TransportRequestOptions.Type.REG));
119119
assertEquals(1, profile.getNumConnectionsPerType(TransportRequestOptions.Type.STATE));
120120
assertEquals(0, profile.getNumConnectionsPerType(TransportRequestOptions.Type.RECOVERY));
121121
assertEquals(3, profile.getNumConnectionsPerType(TransportRequestOptions.Type.BULK));
122122

123-
profile = ConnectionManager.buildDefaultConnectionProfile(Settings.builder().put("node.data", false)
123+
profile = ConnectionProfile.buildDefaultConnectionProfile(Settings.builder().put("node.data", false)
124124
.put("node.master", false).build());
125125
assertEquals(10, profile.getNumConnections());
126126
assertEquals(1, profile.getNumConnectionsPerType(TransportRequestOptions.Type.PING));

‎test/framework/src/main/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java‎

Lines changed: 19 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -116,7 +116,8 @@ public abstract class AbstractSimpleTransportTestCase extends ESTestCase {
116116
protected abstract MockTransportService build(Settings settings, Version version, ClusterSettings clusterSettings, boolean doHandshake);
117117

118118
protected int channelsPerNodeConnection() {
119-
return 13;
119+
// This is a customized profile for this test case.
120+
return 6;
120121
}
121122

122123
@Override
@@ -125,9 +126,17 @@ public void setUp() throws Exception {
125126
super.setUp();
126127
threadPool = new TestThreadPool(getClass().getName());
127128
clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
128-
serviceA = buildService("TS_A", version0, clusterSettings); // this one supports dynamic tracer updates
129+
Settings connectionSettings = Settings.builder()
130+
.put(TransportService.CONNECTIONS_PER_NODE_RECOVERY.getKey(), 1)
131+
.put(TransportService.CONNECTIONS_PER_NODE_BULK.getKey(), 1)
132+
.put(TransportService.CONNECTIONS_PER_NODE_REG.getKey(), 2)
133+
.put(TransportService.CONNECTIONS_PER_NODE_STATE.getKey(), 1)
134+
.put(TransportService.CONNECTIONS_PER_NODE_PING.getKey(), 1)
135+
.build();
136+
137+
serviceA = buildService("TS_A", version0, clusterSettings, connectionSettings); // this one supports dynamic tracer updates
129138
nodeA = serviceA.getLocalNode();
130-
serviceB = buildService("TS_B", version1, null); // this one doesn't support dynamic tracer updates
139+
serviceB = buildService("TS_B", version1, null, connectionSettings); // this one doesn't support dynamic tracer updates
131140
nodeB = serviceB.getLocalNode();
132141
// wait till all nodes are properly connected and the event has been sent, so tests in this class
133142
// will not get this callback called on the connections done in this setup
@@ -174,7 +183,12 @@ private MockTransportService buildService(final String name, final Version versi
174183
}
175184

176185
protected MockTransportService buildService(final String name, final Version version, ClusterSettings clusterSettings) {
177-
return buildService(name, version, clusterSettings, Settings.EMPTY, true, true);
186+
return buildService(name, version, clusterSettings, Settings.EMPTY);
187+
}
188+
189+
protected MockTransportService buildService(final String name, final Version version, ClusterSettings clusterSettings,
190+
Settings settings) {
191+
return buildService(name, version, clusterSettings, settings, true, true);
178192
}
179193

180194
@Override
@@ -1999,7 +2013,7 @@ protected String handleRequest(TcpChannel mockChannel, String profileName, Strea
19992013
assertEquals("handshake failed", exception.getCause().getMessage());
20002014
}
20012015

2002-
ConnectionProfile connectionProfile = ConnectionManager.buildDefaultConnectionProfile(Settings.EMPTY);
2016+
ConnectionProfile connectionProfile = ConnectionProfile.buildDefaultConnectionProfile(Settings.EMPTY);
20032017
try (TransportService service = buildService("TS_TPC", Version.CURRENT, null);
20042018
TcpTransport.NodeChannels connection = originalTransport.openConnection(
20052019
new DiscoveryNode("TS_TPC", "TS_TPC", service.boundAddress().publishAddress(), emptyMap(), emptySet(), version0),

‎test/framework/src/test/java/org/elasticsearch/transport/MockTcpTransportTests.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@ public Version executeHandshake(DiscoveryNode node, TcpChannel mockChannel, Time
5050
}
5151
};
5252
MockTransportService mockTransportService =
53-
MockTransportService.createNewService(Settings.EMPTY, transport, version, threadPool, clusterSettings, Collections.emptySet());
53+
MockTransportService.createNewService(settings, transport, version, threadPool, clusterSettings, Collections.emptySet());
5454
mockTransportService.start();
5555
return mockTransportService;
5656
}

‎test/framework/src/test/java/org/elasticsearch/transport/nio/SimpleMockNioTransportTests.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -78,7 +78,7 @@ protected Version getCurrentVersion() {
7878

7979
};
8080
MockTransportService mockTransportService =
81-
MockTransportService.createNewService(Settings.EMPTY, transport, version, threadPool, clusterSettings, Collections.emptySet());
81+
MockTransportService.createNewService(settings, transport, version, threadPool, clusterSettings, Collections.emptySet());
8282
mockTransportService.start();
8383
return mockTransportService;
8484
}

‎x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/AbstractSimpleSecurityTransportTestCase.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@
2121
import org.elasticsearch.transport.AbstractSimpleTransportTestCase;
2222
import org.elasticsearch.transport.BindTransportException;
2323
import org.elasticsearch.transport.ConnectTransportException;
24-
import org.elasticsearch.transport.ConnectionManager;
2524
import org.elasticsearch.transport.ConnectionProfile;
2625
import org.elasticsearch.transport.TcpTransport;
2726
import org.elasticsearch.transport.TransportRequestOptions;
@@ -111,7 +110,7 @@ public void testTcpHandshake() throws IOException, InterruptedException {
111110
assumeTrue("only tcp transport has a handshake method", serviceA.getOriginalTransport() instanceof TcpTransport);
112111
TcpTransport originalTransport = (TcpTransport) serviceA.getOriginalTransport();
113112

114-
ConnectionProfile connectionProfile = ConnectionManager.buildDefaultConnectionProfile(Settings.EMPTY);
113+
ConnectionProfile connectionProfile = ConnectionProfile.buildDefaultConnectionProfile(Settings.EMPTY);
115114
try (TransportService service = buildService("TS_TPC", Version.CURRENT, null);
116115
TcpTransport.NodeChannels connection = originalTransport.openConnection(
117116
new DiscoveryNode("TS_TPC", "TS_TPC", service.boundAddress().publishAddress(), emptyMap(), emptySet(), version0),

‎x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/netty4/SimpleSecurityNetty4ServerTransportTests.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -93,7 +93,7 @@ protected Version getCurrentVersion() {
9393

9494
};
9595
MockTransportService mockTransportService =
96-
MockTransportService.createNewService(Settings.EMPTY, transport, version, threadPool, clusterSettings,
96+
MockTransportService.createNewService(settings, transport, version, threadPool, clusterSettings,
9797
Collections.emptySet());
9898
mockTransportService.start();
9999
return mockTransportService;

0 commit comments

Comments
 (0)