Add multicast and reduce flakiness

Author: TomaszGaweda

spring/spring-boot-session-replication-spring-session-hazelcast/src/main/java/com/hazelcast/guide/config/SessionConfiguration.java

@@ -43,6 +43,8 @@ public HazelcastInstance hazelcastInstance() {
         Config config = new Config();
         config.setClusterName("spring-session-cluster");
 
+        config.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(true);
+
         // Add this attribute to be able to query sessions by their PRINCIPAL_NAME_ATTRIBUTE's
         AttributeConfig attributeConfig = new AttributeConfig()
                 .setName(HazelcastIndexedSessionRepository.PRINCIPAL_NAME_ATTRIBUTE)

spring/spring-boot-session-replication-spring-session-hazelcast/src/main/java/com/hazelcast/guide/controller/ClusterController.java

@@ -0,0 +1,24 @@
+package com.hazelcast.guide.controller;
+
+import com.hazelcast.core.HazelcastInstance;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * This controller is optional; it's used only to check cluster size to check if cluster already have all nodes connected
+ * in the tests.
+ */
+@RestController
+public class ClusterController {
+
+    private final HazelcastInstance hazelcastInstance;
+
+    public ClusterController(HazelcastInstance hazelcastInstance) {
+        this.hazelcastInstance = hazelcastInstance;
+    }
+
+    @GetMapping("/clusterSize")
+    public int getClusterSize() {
+        return hazelcastInstance.getCluster().getMembers().size();
+    }
+}

spring/spring-boot-session-replication-spring-session-hazelcast/src/main/java/com/hazelcast/guide/controller/SessionController.java

@@ -1,6 +1,7 @@
 package com.hazelcast.guide.controller;
 
 import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
 import jakarta.servlet.http.HttpSession;
 import org.springframework.http.MediaType;
 import org.springframework.session.FindByIndexNameSessionRepository;
@@ -41,7 +42,7 @@ public SessionController(FindByIndexNameSessionRepository<?> sessionRepository)
      *
      */
     @GetMapping("/create")
-    public String createSession(@RequestParam("principal") String principal, HttpServletRequest request) {
+    public String createSession(@RequestParam("principal") String principal, HttpServletRequest request, HttpServletResponse response) {
         HttpSession session = request.getSession(false);
         if (session == null) {
             session = request.getSession();

spring/spring-boot-session-replication-spring-session-hazelcast/src/test/java/com/hazelcast/guide/HazelcastSpringSessionApplicationTests.java

@@ -6,16 +6,20 @@
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.http.HttpEntity;
 import org.springframework.http.HttpHeaders;
-import org.springframework.http.HttpMethod;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.client.RestTemplate;
 import org.springframework.web.util.UriComponentsBuilder;
 
+import java.util.Base64;
 import java.util.Collections;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.springframework.http.HttpMethod.GET;
 
+@SuppressWarnings("DataFlowIssue")
 class HazelcastSpringSessionApplicationTests {
 
 	private static final Logger logger = LoggerFactory.getLogger(HazelcastSpringSessionApplicationTests.class);
@@ -29,16 +33,21 @@ void contextLoads() {
 		String port2 = startApplication();
 		Map<String, String> principalMap = Collections.singletonMap("principal", "hazelcast2020");
 
+		waitForCluster(port1);
+
 		// when
 		ResponseEntity<?> response1 = makeRequest(port1, "create", null, principalMap);
 		String sessionCookie1 = extractSessionCookie(response1);
+		logger.info("First request headers: {}", response1.getHeaders());
+		logger.info("First request body: {}", response1.getBody().toString());
+		logger.info("First session cookie: {}", sessionCookie1);
 
 		// then
 		ResponseEntity<?> response2 = makeRequest(port2, "create", sessionCookie1, principalMap);
 		String body = response2.getBody().toString();
         logger.info("Body contains: {}", body);
-		assertTrue(body.contains("Session already exists"));
 
+		assertThat(body).contains("Session already exists");
 	}
 
 	private static String startApplication() {
@@ -53,25 +62,55 @@ private String extractSessionCookie(ResponseEntity<?> response) {
 		return response.getHeaders().get("Set-Cookie").stream()
 				.filter(s -> s.contains(COOKIE_NAME))
 				.map(s -> s.substring(COOKIE_NAME.length() + 1))
+				.map(s -> s.contains(";") ? s.substring(0, s.indexOf(";")) : s)
+				.map(s -> new String(Base64.getDecoder().decode(s)))
 				.findFirst().orElse(null);
 	}
 
-	private static ResponseEntity<?> makeRequest(String port, String endpoint, String sessionCookie, Map<String, String> parameters) {
+	@SuppressWarnings("SameParameterValue")
+    private static ResponseEntity<?> makeRequest(String port, String endpoint, String sessionCookie, Map<String, String> parameters) {
 		String url = "http://localhost:" + port + "/" + endpoint;
 
 		// Header
 		HttpHeaders headers = new HttpHeaders();
 		if (sessionCookie != null) {
-			headers.add("Cookie", COOKIE_NAME + "=" + sessionCookie);
+			headers.add("Cookie", COOKIE_NAME + "=" +
+					Base64.getEncoder().encodeToString(sessionCookie.getBytes(UTF_8)));
 		}
 
 		// Query parameters
-		UriComponentsBuilder builder = UriComponentsBuilder.fromHttpUrl(url);
+		UriComponentsBuilder builder = UriComponentsBuilder.fromUriString(url);
 		parameters.forEach(builder::queryParam);
 
 		// Request
 		RestTemplate restTemplate = new RestTemplate();
-		return restTemplate.exchange(builder.toUriString(), HttpMethod.GET, new HttpEntity<>(headers), String.class);
+		return restTemplate.exchange(builder.toUriString(), GET, new HttpEntity<>(headers), String.class);
+	}
+
+	public static void waitForCluster(String port) {
+		String url = "http://localhost:" + port + "/clusterSize";
+
+		var restTemplate = new RestTemplate();
+		var requestEntity = new HttpEntity<>(new HttpHeaders());
+
+		long now = System.currentTimeMillis();
+		long timeout = TimeUnit.MINUTES.toMillis(5);
+		int attempts = 0;
+		while (System.currentTimeMillis() - now < timeout) {
+			logger.info("attempt no {}, remaining time {}", attempts + 1, System.currentTimeMillis() - now);
+			ResponseEntity<Integer> clusterSize = restTemplate.exchange(url, GET, requestEntity, Integer.class);
+			if (clusterSize.getBody() == 2) {
+				return;
+			}
+			attempts++;
+			logger.info("Waiting for cluster size, attempt no {} failed", attempts);
+            try {
+                Thread.sleep(1000);
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+        }
+		throw new IllegalStateException("Cluster has incorrect size");
 	}
 
 }