Lock Registry

package org.springframework.integration.jdbc.lock;

public class JdbcLockRegistry
    implements ExpirableLockRegistry<DistributedLock>, RenewableLockRegistry<DistributedLock> {

    public static final Duration DEFAULT_TTL = Duration.ofSeconds(10);

    public JdbcLockRegistry(LockRepository client);
    public JdbcLockRegistry(LockRepository client, Duration expireAfter);

    public void setIdleBetweenTries(Duration idleBetweenTries);
    public void setCacheCapacity(int cacheCapacity);

    public DistributedLock obtain(Object lockKey);
    public void expireUnusedOlderThan(long age);
    public void renewLock(Object lockKey);
    public void renewLock(Object lockKey, Duration customTtl);
}

package org.springframework.integration.jdbc.lock;

public class DefaultLockRepository implements LockRepository, InitializingBean,
        ApplicationContextAware, SmartInitializingSingleton, SmartLifecycle {

    public static final String DEFAULT_TABLE_PREFIX = "INT_";
    public static final Duration DEFAULT_TTL = Duration.ofSeconds(10);

    public DefaultLockRepository(DataSource dataSource);
    public DefaultLockRepository(DataSource dataSource, String id);

    public void setRegion(String region);
    public void setPrefix(String prefix);
    public void setTransactionManager(PlatformTransactionManager transactionManager);
    public void setUpdateQuery(String updateQuery);
    public void setInsertQuery(String insertQuery);
    public void setRenewQuery(String renewQuery);
    public void setCheckDatabaseOnStart(boolean checkDatabaseOnStart);

    public boolean acquire(String lock, Duration ttl);
    public boolean renew(String lock, Duration ttl);
    public boolean isAcquired(String lock);
    public boolean delete(String lock);
    public void deleteExpired();
}

import org.springframework.integration.jdbc.lock.JdbcLockRegistry;
import org.springframework.integration.jdbc.lock.DefaultLockRepository;
import java.util.concurrent.locks.Lock;

// Basic lock registry setup
DefaultLockRepository lockRepository = new DefaultLockRepository(dataSource);
JdbcLockRegistry lockRegistry = new JdbcLockRegistry(lockRepository);

// Obtain and use lock
Lock lock = lockRegistry.obtain("resource-1");
try {
    lock.lock();
    // Critical section - only one JVM can execute this at a time
    System.out.println("Processing shared resource");
    Thread.sleep(1000);
} finally {
    lock.unlock();
}

// Lock with timeout
JdbcLockRegistry timeoutRegistry = new JdbcLockRegistry(
    new DefaultLockRepository(dataSource)
);

Lock timeoutLock = timeoutRegistry.obtain("shared-resource");
if (timeoutLock.tryLock(5, TimeUnit.SECONDS)) {
    try {
        // Got lock within 5 seconds
        performOperation();
    } finally {
        timeoutLock.unlock();
    }
} else {
    // Timeout - couldn't acquire lock
    System.out.println("Failed to acquire lock within timeout");
}

// Renewable locks for long-running operations
JdbcLockRegistry renewableRegistry = new JdbcLockRegistry(
    new DefaultLockRepository(dataSource),
    Duration.ofSeconds(10)
);

Lock renewableLock = renewableRegistry.obtain("batch-job");
renewableLock.lock();
try {
    for (int i = 0; i < 100; i++) {
        // Process batch item
        processBatchItem(i);

        // Renew lock every 10 items to prevent expiration
        if (i % 10 == 0) {
            renewableRegistry.renewLock("batch-job");
        }
    }
} finally {
    renewableLock.unlock();
}

// Expire old unused locks (maintenance task)
JdbcLockRegistry maintenanceRegistry = new JdbcLockRegistry(
    new DefaultLockRepository(dataSource)
);

// Schedule periodic cleanup
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
scheduler.scheduleAtFixedRate(() -> {
    // Expire locks unused for more than 1 hour
    maintenanceRegistry.expireUnusedOlderThan(3600000);
}, 0, 1, TimeUnit.HOURS);

import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.leader.LockRegistryLeaderInitiator;

@Bean
public DefaultLockRepository lockRepository(DataSource dataSource) {
    DefaultLockRepository repository = new DefaultLockRepository(dataSource);
    repository.setRegion("PRODUCTION");
    return repository;
}

@Bean
public JdbcLockRegistry lockRegistry(DefaultLockRepository lockRepository) {
    JdbcLockRegistry registry = new JdbcLockRegistry(lockRepository, Duration.ofSeconds(30));
    registry.setIdleBetweenTries(Duration.ofMillis(100));
    return registry;
}

@Bean
public LockRegistryLeaderInitiator leaderInitiator(JdbcLockRegistry lockRegistry) {
    return new LockRegistryLeaderInitiator(lockRegistry, "cluster-leader");
}

CREATE TABLE INT_LOCK (
    LOCK_KEY CHAR(36) NOT NULL,
    REGION VARCHAR(100) NOT NULL,
    CLIENT_ID CHAR(36),
    CREATED_DATE TIMESTAMP NOT NULL,
    CONSTRAINT INT_LOCK_PK PRIMARY KEY (LOCK_KEY, REGION)
);