@@ -28,26 +28,30 @@ public class SharedClusterManager {
2828 // Increased from default 5 to 500 to support 5,000 collections loading in parallel
2929 private static final int DEFAULT_KV_CONNECTIONS = 5 ;
3030
31- // Shared ClusterEnvironment with optimized connection settings
31+ // Shared ClusterEnvironment for TLS connections
3232 private static ClusterEnvironment sharedEnvironment ;
3333
34- // Track whether environment has been shutdown
35- private static volatile boolean environmentShutdown = false ;
34+ // Shared ClusterEnvironment for non-TLS connections
35+ private static ClusterEnvironment sharedNonTLSEnvironment ;
36+
37+ // Track whether environments have been shutdown
38+ private static volatile boolean tlsEnvironmentShutdown = false ;
39+ private static volatile boolean nonTlsEnvironmentShutdown = false ;
3640
3741 private static final Object environmentLock = new Object ();
3842
3943 // Store cluster instances per server connection string
4044 private static ConcurrentHashMap <String , ClusterWrapper > clusterMap = new ConcurrentHashMap <>();
4145
42- // Initialize the shared environment (lazy initialization with recreation)
46+ // Initialize the shared TLS environment (lazy initialization with recreation)
4347 private static void initializeSharedEnvironment () {
44- if (sharedEnvironment == null || environmentShutdown ) {
48+ if (sharedEnvironment == null || tlsEnvironmentShutdown ) {
4549 synchronized (environmentLock ) {
4650 // Double-check under lock
47- if (sharedEnvironment == null || environmentShutdown ) {
51+ if (sharedEnvironment == null || tlsEnvironmentShutdown ) {
4852 try {
49- if (sharedEnvironment != null && environmentShutdown ) {
50- logger .info ("Shared Cluster Environment was shutdown, recreating" );
53+ if (sharedEnvironment != null && tlsEnvironmentShutdown ) {
54+ logger .info ("Shared TLS Cluster Environment was shutdown, recreating" );
5155 }
5256
5357 sharedEnvironment = ClusterEnvironment .builder ()
@@ -57,10 +61,36 @@ private static void initializeSharedEnvironment() {
5761 .ioConfig (IoConfig .enableDnsSrv (true ))
5862 .ioConfig (IoConfig .numKvConnections (DEFAULT_KV_CONNECTIONS ))
5963 .build ();
60- environmentShutdown = false ;
61- logger .info ("Shared Cluster Environment initialized with " + DEFAULT_KV_CONNECTIONS + " KV connections for massively parallel collection loads " );
64+ tlsEnvironmentShutdown = false ;
65+ logger .info ("Shared TLS Cluster Environment initialized with " + DEFAULT_KV_CONNECTIONS + " KV connections" );
6266 } catch (Exception e ) {
63- logger .error ("Failed to initialize shared Cluster Environment" , e );
67+ logger .error ("Failed to initialize shared TLS Cluster Environment" , e );
68+ }
69+ }
70+ }
71+ }
72+ }
73+
74+ // Initialize the shared non-TLS environment (lazy initialization with recreation)
75+ private static void initializeSharedNonTLSEnvironment () {
76+ if (sharedNonTLSEnvironment == null || nonTlsEnvironmentShutdown ) {
77+ synchronized (environmentLock ) {
78+ // Double-check under lock
79+ if (sharedNonTLSEnvironment == null || nonTlsEnvironmentShutdown ) {
80+ try {
81+ if (sharedNonTLSEnvironment != null && nonTlsEnvironmentShutdown ) {
82+ logger .info ("Shared non-TLS Cluster Environment was shutdown, recreating" );
83+ }
84+
85+ sharedNonTLSEnvironment = ClusterEnvironment .builder ()
86+ .timeoutConfig (TimeoutConfig .builder ().kvTimeout (Duration .ofSeconds (10 )))
87+ .ioConfig (IoConfig .enableDnsSrv (true ))
88+ .ioConfig (IoConfig .numKvConnections (DEFAULT_KV_CONNECTIONS ))
89+ .build ();
90+ nonTlsEnvironmentShutdown = false ;
91+ logger .info ("Shared non-TLS Cluster Environment initialized with " + DEFAULT_KV_CONNECTIONS + " KV connections" );
92+ } catch (Exception e ) {
93+ logger .error ("Failed to initialize shared non-TLS Cluster Environment" , e );
6494 }
6595 }
6696 }
@@ -110,33 +140,54 @@ public static synchronized void releaseCluster(Server server) {
110140 }
111141
112142 /**
113- * Shutdown all cluster instances and the shared environment
143+ * Shutdown all cluster instances and the shared environments
114144 */
115145 public static synchronized void shutdownAll () {
116146 logger .info ("Shutting down all shared Cluster instances" );
117147 for (ClusterWrapper wrapper : clusterMap .values ()) {
118148 if (wrapper .cluster != null ) {
119- wrapper .cluster .disconnect ();
149+ try {
150+ wrapper .cluster .disconnect ();
151+ } catch (Exception e ) {
152+ logger .warn ("Error disconnecting cluster: " + e .getMessage ());
153+ }
120154 }
121155 }
122156 clusterMap .clear ();
123157
124- if (sharedEnvironment != null ) {
125- sharedEnvironment .shutdown ();
126- environmentShutdown = true ;
127- logger .info ("Shared Cluster Environment shutdown complete" );
158+ if (sharedEnvironment != null && !tlsEnvironmentShutdown ) {
159+ try {
160+ sharedEnvironment .shutdown ();
161+ } catch (Exception e ) {
162+ logger .warn ("Error shutting down TLS environment: " + e .getMessage ());
163+ }
164+ tlsEnvironmentShutdown = true ;
165+ logger .info ("Shared TLS Cluster Environment shutdown complete" );
166+ }
167+
168+ if (sharedNonTLSEnvironment != null && !nonTlsEnvironmentShutdown ) {
169+ try {
170+ sharedNonTLSEnvironment .shutdown ();
171+ } catch (Exception e ) {
172+ logger .warn ("Error shutting down non-TLS environment: " + e .getMessage ());
173+ }
174+ nonTlsEnvironmentShutdown = true ;
175+ logger .info ("Shared non-TLS Cluster Environment shutdown complete" );
128176 }
129177 }
130178
131179 private static Cluster createCluster (Server server ) throws AuthenticationFailureException {
180+ initializeSharedEnvironment ();
181+ initializeSharedNonTLSEnvironment ();
182+
132183 ClusterOptions clusterOptions ;
133184 try {
134185 if (server .memcached_port .equals ("11207" )) {
135186 clusterOptions = ClusterOptions .clusterOptions (server .rest_username , server .rest_password )
136187 .environment (sharedEnvironment );
137188 } else {
138189 clusterOptions = ClusterOptions .clusterOptions (server .rest_username , server .rest_password )
139- .environment (createNonTLSEnvironment () );
190+ .environment (sharedNonTLSEnvironment );
140191 }
141192
142193 Cluster cluster = Cluster .connect (server .ip , clusterOptions );
@@ -152,14 +203,6 @@ private static Cluster createCluster(Server server) throws AuthenticationFailure
152203 }
153204 }
154205
155- private static ClusterEnvironment createNonTLSEnvironment () {
156- return ClusterEnvironment .builder ()
157- .timeoutConfig (TimeoutConfig .builder ().kvTimeout (Duration .ofSeconds (10 )))
158- .ioConfig (IoConfig .enableDnsSrv (true ))
159- .ioConfig (IoConfig .numKvConnections (DEFAULT_KV_CONNECTIONS ))
160- .build ();
161- }
162-
163206 private static String getClusterKey (Server server ) {
164207 return server .ip + ":" + server .memcached_port ;
165208 }
![factory-droid[bot]](https://avatars.githubusercontent.com/in/358017?v=4&size=40){kind=avatar}
0 commit comments