feat: micro fix

Author: KirillKurdyukov

shedlock-ydb/src/main/java/tech/ydb/lock/provider/YdbCoordinationServiceLockProvider.java

@@ -11,9 +11,11 @@
 import net.javacrumbs.shedlock.support.Utils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import tech.ydb.common.retry.RetryForever;
 import tech.ydb.coordination.CoordinationClient;
 import tech.ydb.coordination.CoordinationSession;
 import tech.ydb.coordination.SemaphoreLease;
+import tech.ydb.coordination.settings.CoordinationSessionSettings;
 import tech.ydb.core.Result;
 import tech.ydb.jdbc.YdbConnection;
 
@@ -28,8 +30,6 @@ public class YdbCoordinationServiceLockProvider implements LockProvider {
     private final YdbConnection ydbConnection;
     private final CoordinationClient coordinationClient;
 
-    private volatile CoordinationSession coordinationSession;
-
     public YdbCoordinationServiceLockProvider(YdbConnection ydbConnection) {
         this.ydbConnection = ydbConnection;
         this.coordinationClient = CoordinationClient.newClient(ydbConnection.getCtx().getGrpcTransport());
@@ -39,21 +39,6 @@ public void init() {
         for (int i = 0; i < ATTEMPT_CREATE_NODE; i++) {
             var status = coordinationClient.createNode(YDB_LOCK_NODE_NAME).join();
 
-            if (status.isSuccess()) {
-                coordinationSession = coordinationClient.createSession(YDB_LOCK_NODE_NAME);
-
-                var statusCS = coordinationSession.connect().join();
-
-                if (statusCS.isSuccess()) {
-                    logger.info("Created coordination node session [{}]", coordinationSession);
-
-                    return;
-                }
-                if (i == ATTEMPT_CREATE_NODE - 1) {
-                    statusCS.expectSuccess("Failed creating coordination node session");
-                }
-            }
-
             if (i == ATTEMPT_CREATE_NODE - 1) {
                 status.expectSuccess("Failed created coordination service node: " + YDB_LOCK_NODE_NAME);
             }
@@ -70,39 +55,57 @@ public Optional<SimpleLock> lock(LockConfiguration lockConfiguration) {
 
         logger.info("Instance[{}] is trying to become a leader...", instanceInfo);
 
+        var coordinationSession = coordinationClient.createSession(
+                YDB_LOCK_NODE_NAME, CoordinationSessionSettings.newBuilder()
+                        .withRetryPolicy(new RetryForever(500))
+                        .build()
+        );
+
+        var statusCS = coordinationSession.connect().join();
+
+        if (!statusCS.isSuccess()) {
+            logger.info("Failed creating coordination session [{}]", coordinationSession);
+
+            return Optional.empty();
+        }
+
+        logger.info("Created coordination node session [{}]", coordinationSession);
+
         Result<SemaphoreLease> semaphoreLease = coordinationSession.acquireEphemeralSemaphore(
                 lockConfiguration.getName(),
                 true,
                 instanceInfo.getBytes(StandardCharsets.UTF_8),
                 lockConfiguration.getLockAtMostFor()
         ).join();
 
+        logger.debug(coordinationSession.toString());
+
         if (semaphoreLease.isSuccess()) {
             logger.info("Instance[{}] acquired semaphore[SemaphoreName={}]", instanceInfo,
                     semaphoreLease.getValue().getSemaphoreName());
 
-            return Optional.of(new YdbSimpleLock(semaphoreLease.getValue(), instanceInfo));
+            return Optional.of(new YdbSimpleLock(semaphoreLease.getValue(), instanceInfo, coordinationSession));
         } else {
             logger.info("Instance[{}] did not acquire semaphore", instanceInfo);
 
             return Optional.empty();
         }
     }
 
-    private record YdbSimpleLock(SemaphoreLease semaphoreLease, String metaInfo) implements SimpleLock {
+    private record YdbSimpleLock(SemaphoreLease semaphoreLease, String metaInfo,
+                                 CoordinationSession coordinationSession) implements SimpleLock {
         @Override
         public void unlock() {
             logger.info("Instance[{}] released semaphore[SemaphoreName={}]", metaInfo, semaphoreLease.getSemaphoreName());
 
             semaphoreLease.release().join();
+
+            coordinationSession.close();
         }
     }
 
     @PreDestroy
     private void close() throws SQLException {
-        // closing coordination session
-        coordinationSession.close();
-
         ydbConnection.close();
     }
 }