lingwei.kong/hawkbit
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

Clean up distribution lock (#3081)

Browse Source 
Signed-off-by: Avgustin Marinov <Avgustin.Marinov@bosch.com>
This commit is contained in:
Avgustin Marinov
2026-05-14 09:10:17 +03:00
committed by GitHub
parent 275348d524
commit 6311e64ea9
3 changed files with 37 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/JpaRepositoryConfiguration.java
View File

@@ -202,9 +202,8 @@ public class JpaRepositoryConfiguration {
@Bean @Bean
@ConditionalOnProperty(name = "hawkbit.lock", havingValue = "distributed", matchIfMissing = true) @ConditionalOnProperty(name = "hawkbit.lock", havingValue = "distributed", matchIfMissing = true)
@ConditionalOnMissingBean @ConditionalOnMissingBean
LockRepository lockRepository(final DataSource dataSource, final LockProperties lockProperties, LockRepository lockRepository(final DataSource dataSource, final LockProperties lockProperties) {
final PlatformTransactionManager txManager) { final DefaultLockRepository repository = new DistributedLockRepository(dataSource, lockProperties);
final DefaultLockRepository repository = new DistributedLockRepository(dataSource, lockProperties, txManager);
repository.setPrefix("SP_"); repository.setPrefix("SP_");
return repository; return repository;
} }

94
hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/cluster/DistributedLockRepository.java
View File

@@ -19,24 +19,21 @@ import java.util.concurrent.ConcurrentHashMap;
import javax.sql.DataSource; import javax.sql.DataSource;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.eclipse.hawkbit.repository.jpa.utils.DeploymentHelper; import org.jspecify.annotations.NullMarked;
import org.springframework.dao.DataIntegrityViolationException; import org.jspecify.annotations.Nullable;
import org.springframework.dao.PessimisticLockingFailureException; import org.springframework.dao.PessimisticLockingFailureException;
import org.springframework.dao.QueryTimeoutException;
import org.springframework.integration.jdbc.lock.DefaultLockRepository; import org.springframework.integration.jdbc.lock.DefaultLockRepository;
import org.springframework.integration.jdbc.lock.JdbcLockRegistry; import org.springframework.integration.jdbc.lock.JdbcLockRegistry;
import org.springframework.scheduling.annotation.Scheduled; import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionTimedOutException;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
/** /**
* Repository for {@link JdbcLockRegistry}. This class is not thread safe. Adds support for keeping lock longer then ttl * Repository for {@link JdbcLockRegistry}. This class is not thread safe.
* if really used by the instance. * Adds support for keeping lock longer than ttl if really used by the instance (renew).
*/ */
@Slf4j @Slf4j
@NullMarked
public class DistributedLockRepository extends DefaultLockRepository { public class DistributedLockRepository extends DefaultLockRepository {
private static final int MAX_DELETE_RETRY = 10; private static final int MAX_DELETE_RETRY = 10;
@@ -44,35 +41,32 @@ public class DistributedLockRepository extends DefaultLockRepository {
// period between successive refresh tics // period between successive refresh tics
private static final String TIC_PERIOD_MS = "${hawkbit.repository.cluster.lock.ticPeriodMS:2000}"; private static final String TIC_PERIOD_MS = "${hawkbit.repository.cluster.lock.ticPeriodMS:2000}";
private final PlatformTransactionManager txManager; private final Duration renewTtl;
private final int refreshOnRemainMS; // if null refresh is effectively disabled (when both REFRESH_ON_REMAINS_MS and REFRESH_ON_REMAINS_PERCENT are non-positive)
private final int refreshOnRemainPercent;
// if empty refresh is effectively disabled (when both REFRESH_ON_REMAINS_MS and REFRESH_ON_REMAINS_PERCENT are non-positive)
// otherwise, a refresh is triggered at refreshAfterMillis after lock acquisition or last refresh // otherwise, a refresh is triggered at refreshAfterMillis after lock acquisition or last refresh
private Optional<Integer> refreshAfterMillis; @Nullable
private final Integer refreshAfterMillis;
// lock <-> next refresh time // lock <-> next refresh time
private final Map<String, Instant> lockToRefreshTime = new ConcurrentHashMap<>(); private final Map<String, Instant> lockToRefreshTime = new ConcurrentHashMap<>();
public DistributedLockRepository( public DistributedLockRepository(final DataSource dataSource, final LockProperties lockProperties) {
final DataSource dataSource, final LockProperties lockProperties, final PlatformTransactionManager txManager) {
super(dataSource); super(dataSource);
this.txManager = txManager;
this.refreshOnRemainMS = lockProperties.getRefreshOnRemainMS(); renewTtl = Duration.ofMillis(lockProperties.getTtl());
this.refreshOnRemainPercent = lockProperties.getRefreshOnRemainPercent();
final int timeToLive = lockProperties.getTtl();
final int triggerOnRemainMS = Math.max(
lockProperties.getRefreshOnRemainMS(),
timeToLive * lockProperties.getRefreshOnRemainPercent() / 100);
final int refreshAfterMS = timeToLive - triggerOnRemainMS;
refreshAfterMillis = refreshAfterMS <= 0 ? null : refreshAfterMS;
// to ensure that deprecated acquire and renew will use the lockProperties ttl. none shall call them but anyway - for sure
// to be removed when the #setTimeToLive method is removed from DefaultLockRepository
setTimeToLive(lockProperties.getTtl()); setTimeToLive(lockProperties.getTtl());
} }
// interceptor that handles refreshAfterMillis update when time to live is changed
@Override
public void setTimeToLive(final int timeToLive) {
super.setTimeToLive(timeToLive);
refreshAfterMillis = refreshAfterMillis(timeToLive);
}
@Transactional(propagation = Propagation.REQUIRES_NEW) @Transactional(propagation = Propagation.REQUIRES_NEW)
@Override @Override
public boolean delete(final String lock) { public boolean delete(final String lock) {
@@ -98,30 +92,20 @@ public class DistributedLockRepository extends DefaultLockRepository {
// Spring Integration 7.0 calls acquire(String, Duration) directly; the deprecated acquire(String) is no longer invoked // Spring Integration 7.0 calls acquire(String, Duration) directly; the deprecated acquire(String) is no longer invoked
// by JdbcLockRegistry. Override the new method to populate lockToRefreshTime for the refresh mechanism. // by JdbcLockRegistry. Override the new method to populate lockToRefreshTime for the refresh mechanism.
@Transactional(propagation = Propagation.NOT_SUPPORTED)
@Override @Override
public boolean acquire(final String lock, final Duration ttl) { public boolean acquire(final String lock, final Duration ttl) {
try { final boolean acquired = super.acquire(lock, ttl);
// run in a new transaction so we know the committed result before updating lockToRefreshTime if (acquired) {
final boolean acquired = DeploymentHelper.runInNewTransaction( Optional.ofNullable(refreshAfterMillis).ifPresent(afterMillis ->
txManager, "lock-acquire", Isolation.READ_COMMITTED.value(), status -> super.acquire(lock, ttl)); lockToRefreshTime.put(lock, Instant.now().plus(afterMillis, ChronoUnit.MILLIS)));
if (acquired) {
refreshAfterMillis.ifPresent(afterMillis -> lockToRefreshTime.put(lock, Instant.now().plus(afterMillis, ChronoUnit.MILLIS)));
}
return acquired;
} catch (final DataIntegrityViolationException | PessimisticLockingFailureException e) {
log.debug("Could not acquire cluster lock {}. I guess another node has it.", lock, e);
return false;
} catch (final QueryTimeoutException e) {
log.debug("Query timed out for lock {}.", lock, e);
throw new TransactionTimedOutException("DB query timed out for lock " + lock, e);
} }
return acquired;
} }
@SuppressWarnings({ "java:S1066" }) @SuppressWarnings({ "java:S1066" })
@Scheduled(initialDelayString = TIC_PERIOD_MS, fixedDelayString = TIC_PERIOD_MS) @Scheduled(initialDelayString = TIC_PERIOD_MS, fixedDelayString = TIC_PERIOD_MS)
public void refresh() { public void renew() {
refreshAfterMillis.ifPresentOrElse(afterMillis -> { Optional.ofNullable(refreshAfterMillis).ifPresentOrElse(afterMillis -> {
final Instant now = Instant.now(); final Instant now = Instant.now();
lockToRefreshTime.forEach((lock, refreshTime) -> { lockToRefreshTime.forEach((lock, refreshTime) -> {
if (now.isAfter(refreshTime)) { if (now.isAfter(refreshTime)) {
@@ -129,8 +113,8 @@ public class DistributedLockRepository extends DefaultLockRepository {
// if delete is called while iterating we must skip record update // if delete is called while iterating we must skip record update
// otherwise, the lock will be unavailable for everyone until expiration // otherwise, the lock will be unavailable for everyone until expiration
if (lockToRefreshTime.containsKey(lock)) { if (lockToRefreshTime.containsKey(lock)) {
if (!acquire(lock)) { // try to update record in lock table if (!renew(lock, renewTtl)) { // try to update record in lock table
log.warn("Failed to refresh cluster lock {}!", lock); log.warn("Failed to renew cluster lock {}!", lock);
} }
} }
} }
@@ -138,24 +122,4 @@ public class DistributedLockRepository extends DefaultLockRepository {
}); });
}, lockToRefreshTime::clear); }, lockToRefreshTime::clear);
} }
private Optional<Integer> refreshAfterMillis(final int timeToLive) {
final int triggerOnRemainMS = Math.max(refreshOnRemainMS, timeToLive * refreshOnRemainPercent / 100);
final int refreshAfterMS = timeToLive - triggerOnRemainMS;
return refreshAfterMS <= 0 ? Optional.empty() : Optional.of(refreshAfterMS);
}
// * May be required if the super class doesn't execute in new transactions
// * See https://github.com/spring-projects/spring-integration/issues/3683
// * may be not needed anymore
// @Transactional(propagation = Propagation.REQUIRES_NEW, readOnly = true)
// @Override
// public boolean isAcquired(final String lock) {
// return super.isAcquired(lock);
// }
//
// @Transactional(propagation = Propagation.REQUIRES_NEW)
// @Override
// public void close() {
// super.close();
// }
} }

16
hawkbit-repository/hawkbit-repository-jpa/src/test/java/org/eclipse/hawkbit/repository/jpa/cluster/DistributedLockTest.java
View File

@@ -37,7 +37,6 @@ import org.springframework.integration.support.locks.LockRegistry;
import org.springframework.integration.util.UUIDConverter; import org.springframework.integration.util.UUIDConverter;
import org.springframework.security.core.userdetails.UserDetailsService; import org.springframework.security.core.userdetails.UserDetailsService;
import org.springframework.security.provisioning.InMemoryUserDetailsManager; import org.springframework.security.provisioning.InMemoryUserDetailsManager;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.annotation.EnableTransactionManagement; import org.springframework.transaction.annotation.EnableTransactionManagement;
/** /**
@@ -77,20 +76,17 @@ class DistributedLockTest extends AbstractJpaIntegrationTest {
} }
@Bean @Bean
LockRepository lockRepository0(final DataSource dataSource, final LockProperties lockProperties, LockRepository lockRepository0(final DataSource dataSource, final LockProperties lockProperties) {
final PlatformTransactionManager txManager) { return lockRepository(dataSource, lockProperties);
return lockRepository(dataSource, lockProperties, txManager);
} }
@Bean @Bean
LockRepository lockRepository1(final DataSource dataSource, final LockProperties lockProperties, LockRepository lockRepository1(final DataSource dataSource, final LockProperties lockProperties) {
final PlatformTransactionManager txManager) { return lockRepository(dataSource, lockProperties);
return lockRepository(dataSource, lockProperties, txManager);
} }
private LockRepository lockRepository(final DataSource dataSource, final LockProperties lockProperties, private LockRepository lockRepository(final DataSource dataSource, final LockProperties lockProperties) {
final PlatformTransactionManager txManager) { final DefaultLockRepository repository = new DistributedLockRepository(dataSource, lockProperties);
final DefaultLockRepository repository = new DistributedLockRepository(dataSource, lockProperties, txManager);
repository.setPrefix("SP_"); repository.setPrefix("SP_");
return repository; return repository;
} }