mirror of
https://github.com/dromara/hertzbeat.git
synced 2026-09-17 09:40:58 +00:00
Coordinate metadata migration maintenance
This commit is contained in:
Logic
+68
@@ -0,0 +1,68 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.manager.maintenance;
|
||||
|
||||
import org.apache.hertzbeat.common.transaction.MetadataWriteMaintenanceLease;
|
||||
|
||||
/** Releases one acquired maintenance window in strict reverse order with retryable progress. */
|
||||
final class CompositeMigrationMaintenanceLease implements MigrationMaintenanceLease {
|
||||
|
||||
private final DeploymentSingletonLease authorityLease;
|
||||
private final MigrationSourceLease sourceLease;
|
||||
private final MetadataMaintenanceLease producerLease;
|
||||
private final MetadataWriteMaintenanceLease writeLease;
|
||||
private final Runnable reservationRelease;
|
||||
private boolean writeReleased;
|
||||
private boolean producerReleased;
|
||||
private boolean sourceReleased;
|
||||
private boolean authorityReleased;
|
||||
|
||||
CompositeMigrationMaintenanceLease(
|
||||
DeploymentSingletonLease authorityLease,
|
||||
MigrationSourceLease sourceLease,
|
||||
MetadataMaintenanceLease producerLease,
|
||||
MetadataWriteMaintenanceLease writeLease,
|
||||
Runnable reservationRelease) {
|
||||
this.authorityLease = authorityLease;
|
||||
this.sourceLease = sourceLease;
|
||||
this.producerLease = producerLease;
|
||||
this.writeLease = writeLease;
|
||||
this.reservationRelease = reservationRelease;
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void close() {
|
||||
try {
|
||||
releaseInOrder();
|
||||
} catch (MigrationMaintenanceException exception) {
|
||||
throw exception;
|
||||
} catch (RuntimeException exception) {
|
||||
throw MigrationMaintenanceException.resumeFailure();
|
||||
}
|
||||
reservationRelease.run();
|
||||
}
|
||||
|
||||
private void releaseInOrder() {
|
||||
if (!writeReleased) {
|
||||
writeLease.close();
|
||||
writeReleased = true;
|
||||
}
|
||||
if (!producerReleased) {
|
||||
producerLease.resume();
|
||||
producerReleased = true;
|
||||
}
|
||||
if (!sourceReleased) {
|
||||
sourceLease.close();
|
||||
sourceReleased = true;
|
||||
}
|
||||
if (!authorityReleased) {
|
||||
authorityLease.close();
|
||||
authorityReleased = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
+152
@@ -0,0 +1,152 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.manager.maintenance;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.sql.SQLException;
|
||||
import java.time.Duration;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.RejectedExecutionException;
|
||||
import java.util.concurrent.SynchronousQueue;
|
||||
import java.util.concurrent.ThreadPoolExecutor;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import javax.sql.DataSource;
|
||||
|
||||
/** Bounded JDBC connection acquisition that closes every result arriving after its deadline. */
|
||||
final class DeadlineConnectionAcquirer implements AutoCloseable {
|
||||
|
||||
private final DataSource dataSource;
|
||||
private final ThreadPoolExecutor executor;
|
||||
|
||||
DeadlineConnectionAcquirer(DataSource dataSource) {
|
||||
this.dataSource = dataSource;
|
||||
executor = new ThreadPoolExecutor(0, 1, 30, TimeUnit.SECONDS, new SynchronousQueue<>(),
|
||||
Thread.ofPlatform().daemon(true).name("migration-source-connection", 0).factory());
|
||||
}
|
||||
|
||||
Connection acquire(Duration timeout) {
|
||||
long timeoutNanos;
|
||||
try {
|
||||
timeoutNanos = timeout.toNanos();
|
||||
} catch (ArithmeticException exception) {
|
||||
throw MigrationMaintenanceException.invalidRequest();
|
||||
}
|
||||
Attempt attempt = new Attempt();
|
||||
try {
|
||||
executor.execute(() -> connect(attempt));
|
||||
} catch (RejectedExecutionException exception) {
|
||||
if (executor.isShutdown()) {
|
||||
throw MigrationMaintenanceException.sourceUnavailable();
|
||||
}
|
||||
throw MigrationMaintenanceException.timeout();
|
||||
}
|
||||
return attempt.await(timeoutNanos);
|
||||
}
|
||||
|
||||
private void connect(Attempt attempt) {
|
||||
Connection acquired = null;
|
||||
Throwable failure = null;
|
||||
try {
|
||||
acquired = dataSource.getConnection();
|
||||
} catch (Throwable connectionFailure) {
|
||||
failure = connectionFailure;
|
||||
}
|
||||
attempt.complete(acquired, failure);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
executor.shutdownNow();
|
||||
}
|
||||
|
||||
private static final class Attempt {
|
||||
|
||||
private final Object lock = new Object();
|
||||
private final CountDownLatch completed = new CountDownLatch(1);
|
||||
private Connection connection;
|
||||
private Throwable failure;
|
||||
private boolean abandoned;
|
||||
private boolean finished;
|
||||
|
||||
private Connection await(long timeoutNanos) {
|
||||
try {
|
||||
if (completed.await(timeoutNanos, TimeUnit.NANOSECONDS)) {
|
||||
return claimCompleted();
|
||||
}
|
||||
return abandonOrClaim();
|
||||
} catch (InterruptedException exception) {
|
||||
abandon();
|
||||
Thread.currentThread().interrupt();
|
||||
throw MigrationMaintenanceException.interrupted();
|
||||
}
|
||||
}
|
||||
|
||||
private void complete(Connection acquired, Throwable acquiredFailure) {
|
||||
synchronized (lock) {
|
||||
if (abandoned) {
|
||||
closeLate(acquired);
|
||||
} else {
|
||||
connection = acquired;
|
||||
failure = acquiredFailure;
|
||||
}
|
||||
finished = true;
|
||||
}
|
||||
completed.countDown();
|
||||
}
|
||||
|
||||
private Connection abandonOrClaim() {
|
||||
synchronized (lock) {
|
||||
if (finished) {
|
||||
return claimCompletedLocked();
|
||||
}
|
||||
abandoned = true;
|
||||
}
|
||||
throw MigrationMaintenanceException.timeout();
|
||||
}
|
||||
|
||||
private void abandon() {
|
||||
synchronized (lock) {
|
||||
if (!finished) {
|
||||
abandoned = true;
|
||||
} else {
|
||||
closeLate(connection);
|
||||
connection = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private Connection claimCompleted() {
|
||||
synchronized (lock) {
|
||||
return claimCompletedLocked();
|
||||
}
|
||||
}
|
||||
|
||||
private Connection claimCompletedLocked() {
|
||||
if (failure instanceof Error error) {
|
||||
throw error;
|
||||
}
|
||||
if (failure != null || connection == null) {
|
||||
throw MigrationMaintenanceException.sourceUnavailable();
|
||||
}
|
||||
Connection claimed = connection;
|
||||
connection = null;
|
||||
return claimed;
|
||||
}
|
||||
|
||||
private static void closeLate(Connection lateConnection) {
|
||||
if (lateConnection == null) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
lateConnection.close();
|
||||
} catch (SQLException | RuntimeException exception) {
|
||||
// A late connection never becomes a lease; its details remain private.
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+246
@@ -0,0 +1,246 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.manager.maintenance;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
import java.util.function.LongSupplier;
|
||||
import org.apache.hertzbeat.common.transaction.MetadataWriteAdmissionCoordinator;
|
||||
import org.apache.hertzbeat.common.transaction.MetadataWriteAdmissionErrorCode;
|
||||
import org.apache.hertzbeat.common.transaction.MetadataWriteAdmissionException;
|
||||
import org.apache.hertzbeat.common.transaction.MetadataWriteMaintenanceLease;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/** Composes deployment, source, producer, and transaction fences without owning their state machines. */
|
||||
@Component
|
||||
public final class DefaultMigrationMaintenanceOrchestrator implements MigrationMaintenanceOrchestrator {
|
||||
|
||||
private final ReentrantLock lock = new ReentrantLock();
|
||||
private final DeploymentSingletonAuthority deploymentAuthority;
|
||||
private final MigrationSourceGuard sourceGuard;
|
||||
private final MetadataMaintenanceCoordinator producerCoordinator;
|
||||
private final MetadataWriteAdmissionCoordinator writeCoordinator;
|
||||
private final LongSupplier ticker;
|
||||
private String operationId;
|
||||
private Object ownerToken;
|
||||
private MetadataMaintenanceLease recoveryProducerLease;
|
||||
private boolean recoveryProducerCoordinatorRequired;
|
||||
private MigrationSourceLease recoverySourceLease;
|
||||
private DeploymentSingletonLease recoveryAuthorityLease;
|
||||
|
||||
@Autowired
|
||||
public DefaultMigrationMaintenanceOrchestrator(
|
||||
DeploymentSingletonAuthority deploymentAuthority,
|
||||
MigrationSourceGuard sourceGuard,
|
||||
MetadataMaintenanceCoordinator producerCoordinator,
|
||||
MetadataWriteAdmissionCoordinator writeCoordinator) {
|
||||
this(deploymentAuthority, sourceGuard, producerCoordinator, writeCoordinator, System::nanoTime);
|
||||
}
|
||||
|
||||
DefaultMigrationMaintenanceOrchestrator(
|
||||
DeploymentSingletonAuthority deploymentAuthority,
|
||||
MigrationSourceGuard sourceGuard,
|
||||
MetadataMaintenanceCoordinator producerCoordinator,
|
||||
MetadataWriteAdmissionCoordinator writeCoordinator,
|
||||
LongSupplier ticker) {
|
||||
this.deploymentAuthority = deploymentAuthority;
|
||||
this.sourceGuard = sourceGuard;
|
||||
this.producerCoordinator = producerCoordinator;
|
||||
this.writeCoordinator = writeCoordinator;
|
||||
this.ticker = ticker;
|
||||
}
|
||||
|
||||
@Override
|
||||
public MigrationMaintenanceLease acquire(String requestedOperationId, Duration timeout) {
|
||||
requireRequest(requestedOperationId);
|
||||
MaintenanceDeadline deadline;
|
||||
try {
|
||||
deadline = MaintenanceDeadline.start(timeout, ticker);
|
||||
} catch (MetadataMaintenanceException exception) {
|
||||
throw MigrationMaintenanceException.invalidRequest();
|
||||
}
|
||||
Object token = reserve(requestedOperationId);
|
||||
DeploymentSingletonLease authorityLease = null;
|
||||
MigrationSourceLease sourceLease = null;
|
||||
MetadataMaintenanceLease producerLease = null;
|
||||
try {
|
||||
authorityLease = deploymentAuthority.acquire(requestedOperationId, deadline.remaining());
|
||||
sourceLease = sourceGuard.fence(requestedOperationId, deadline.remaining());
|
||||
producerLease = producerCoordinator.quiesce(requestedOperationId, deadline.remaining());
|
||||
MetadataWriteMaintenanceLease writeLease =
|
||||
writeCoordinator.acquire(requestedOperationId, deadline.remaining());
|
||||
return new CompositeMigrationMaintenanceLease(
|
||||
authorityLease, sourceLease, producerLease, writeLease, () -> releaseReservation(token));
|
||||
} catch (Error error) {
|
||||
if (cleanupFailedAcquisition(error, producerLease, sourceLease, authorityLease)) {
|
||||
releaseReservation(token);
|
||||
} else {
|
||||
retainRecovery(producerLease, sourceLease, authorityLease);
|
||||
}
|
||||
throw error;
|
||||
} catch (RuntimeException exception) {
|
||||
MigrationMaintenanceException primary = mapFailure(exception);
|
||||
if (cleanupFailedAcquisition(primary, producerLease, sourceLease, authorityLease)) {
|
||||
releaseReservation(token);
|
||||
} else {
|
||||
retainRecovery(producerLease, sourceLease, authorityLease);
|
||||
}
|
||||
throw primary;
|
||||
}
|
||||
}
|
||||
|
||||
private Object reserve(String requestedOperationId) {
|
||||
lock.lock();
|
||||
try {
|
||||
if (ownerToken != null) {
|
||||
if (!requestedOperationId.equals(operationId)
|
||||
|| !hasRecovery()) {
|
||||
throw MigrationMaintenanceException.operationConflict();
|
||||
}
|
||||
recoverFailedAcquisition();
|
||||
}
|
||||
Object token = new Object();
|
||||
ownerToken = token;
|
||||
operationId = requestedOperationId;
|
||||
return token;
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
private boolean cleanupFailedAcquisition(
|
||||
Throwable primary,
|
||||
MetadataMaintenanceLease producerLease,
|
||||
MigrationSourceLease sourceLease,
|
||||
DeploymentSingletonLease authorityLease) {
|
||||
boolean interrupted = Thread.currentThread().isInterrupted();
|
||||
boolean producerReleased = producerCoordinator.snapshot().phase() == MetadataMaintenancePhase.RUNNING;
|
||||
if (producerLease != null) {
|
||||
producerReleased = suppressCleanup(primary, producerLease::resume);
|
||||
}
|
||||
boolean sourceReleased = sourceLease == null;
|
||||
if (producerReleased && sourceLease != null) {
|
||||
sourceReleased = suppressCleanup(primary, sourceLease::close);
|
||||
}
|
||||
boolean authorityReleased = authorityLease == null;
|
||||
if (producerReleased && sourceReleased && authorityLease != null) {
|
||||
authorityReleased = suppressCleanup(primary, authorityLease::close);
|
||||
}
|
||||
if (interrupted) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
return producerReleased && sourceReleased && authorityReleased;
|
||||
}
|
||||
|
||||
private boolean suppressCleanup(Throwable primary, Runnable cleanup) {
|
||||
try {
|
||||
cleanup.run();
|
||||
return true;
|
||||
} catch (RuntimeException exception) {
|
||||
primary.addSuppressed(MigrationMaintenanceException.resumeFailure());
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private void retainRecovery(
|
||||
MetadataMaintenanceLease producerLease,
|
||||
MigrationSourceLease sourceLease,
|
||||
DeploymentSingletonLease authorityLease) {
|
||||
lock.lock();
|
||||
try {
|
||||
recoveryProducerLease = producerLease;
|
||||
recoveryProducerCoordinatorRequired = producerLease == null
|
||||
&& producerCoordinator.snapshot().phase() != MetadataMaintenancePhase.RUNNING;
|
||||
recoverySourceLease = sourceLease;
|
||||
recoveryAuthorityLease = authorityLease;
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
private void recoverFailedAcquisition() {
|
||||
try {
|
||||
if (recoveryProducerLease != null) {
|
||||
recoveryProducerLease.resume();
|
||||
recoveryProducerLease = null;
|
||||
}
|
||||
if (recoveryProducerCoordinatorRequired) {
|
||||
producerCoordinator.recover(operationId);
|
||||
recoveryProducerCoordinatorRequired = false;
|
||||
}
|
||||
if (recoverySourceLease != null) {
|
||||
recoverySourceLease.close();
|
||||
recoverySourceLease = null;
|
||||
}
|
||||
if (recoveryAuthorityLease != null) {
|
||||
recoveryAuthorityLease.close();
|
||||
recoveryAuthorityLease = null;
|
||||
}
|
||||
ownerToken = null;
|
||||
operationId = null;
|
||||
} catch (RuntimeException exception) {
|
||||
throw MigrationMaintenanceException.resumeFailure();
|
||||
}
|
||||
}
|
||||
|
||||
private boolean hasRecovery() {
|
||||
return recoveryProducerLease != null
|
||||
|| recoveryProducerCoordinatorRequired
|
||||
|| recoverySourceLease != null
|
||||
|| recoveryAuthorityLease != null;
|
||||
}
|
||||
|
||||
private MigrationMaintenanceException mapFailure(RuntimeException exception) {
|
||||
if (exception instanceof MigrationMaintenanceException migrationFailure) {
|
||||
return migrationFailure;
|
||||
}
|
||||
if (exception instanceof MetadataMaintenanceException maintenanceFailure) {
|
||||
return switch (maintenanceFailure.code()) {
|
||||
case INVALID_REQUEST -> MigrationMaintenanceException.invalidRequest();
|
||||
case OPERATION_CONFLICT, STALE_LEASE -> MigrationMaintenanceException.operationConflict();
|
||||
case QUIESCE_TIMEOUT -> MigrationMaintenanceException.timeout();
|
||||
case QUIESCE_INTERRUPTED -> MigrationMaintenanceException.interrupted();
|
||||
case PARTICIPANT_FAILURE, RESUME_FAILURE -> MigrationMaintenanceException.maintenanceFailure();
|
||||
};
|
||||
}
|
||||
if (exception instanceof MetadataWriteAdmissionException writeFailure) {
|
||||
MetadataWriteAdmissionErrorCode code = writeFailure.code();
|
||||
return switch (code) {
|
||||
case INVALID_REQUEST -> MigrationMaintenanceException.invalidRequest();
|
||||
case OPERATION_CONFLICT, MAINTENANCE_ACTIVE -> MigrationMaintenanceException.operationConflict();
|
||||
case DRAIN_TIMEOUT -> MigrationMaintenanceException.timeout();
|
||||
case ACQUISITION_INTERRUPTED -> MigrationMaintenanceException.interrupted();
|
||||
};
|
||||
}
|
||||
return MigrationMaintenanceException.maintenanceFailure();
|
||||
}
|
||||
|
||||
private void releaseReservation(Object token) {
|
||||
lock.lock();
|
||||
try {
|
||||
if (ownerToken == token) {
|
||||
ownerToken = null;
|
||||
operationId = null;
|
||||
recoveryProducerLease = null;
|
||||
recoveryProducerCoordinatorRequired = false;
|
||||
recoverySourceLease = null;
|
||||
recoveryAuthorityLease = null;
|
||||
}
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
private void requireRequest(String requestedOperationId) {
|
||||
if (requestedOperationId == null || requestedOperationId.isBlank()) {
|
||||
throw MigrationMaintenanceException.invalidRequest();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
+16
@@ -0,0 +1,16 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.manager.maintenance;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
/** Proves and fences single-manager ownership for one logical deployment. */
|
||||
public interface DeploymentSingletonAuthority {
|
||||
|
||||
DeploymentSingletonLease acquire(String operationId, Duration timeout);
|
||||
}
|
||||
+15
@@ -0,0 +1,15 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.manager.maintenance;
|
||||
|
||||
/** Owner capability for one authoritative deployment-singleton fence. */
|
||||
public interface DeploymentSingletonLease extends AutoCloseable {
|
||||
|
||||
@Override
|
||||
void close();
|
||||
}
|
||||
+98
@@ -0,0 +1,98 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.manager.maintenance;
|
||||
|
||||
import java.nio.file.Path;
|
||||
import java.util.Locale;
|
||||
|
||||
/** Pure embedded-H2 source-safety classification. */
|
||||
final class EmbeddedH2SourceClassifier {
|
||||
|
||||
private static final String H2_PREFIX = "jdbc:h2:";
|
||||
|
||||
private EmbeddedH2SourceClassifier() {
|
||||
}
|
||||
|
||||
static boolean isSafeEmbeddedSource(String productName, String jdbcUrl) {
|
||||
if (productName == null || !"h2".equals(productName.trim().toLowerCase(Locale.ROOT))
|
||||
|| jdbcUrl == null) {
|
||||
return false;
|
||||
}
|
||||
String url = jdbcUrl.trim().toLowerCase(Locale.ROOT);
|
||||
if (!url.startsWith(H2_PREFIX) || hasUnsafeSetting(url)) {
|
||||
return false;
|
||||
}
|
||||
String location = url.substring(H2_PREFIX.length());
|
||||
if (location.startsWith("mem:")) {
|
||||
return !isRemote(location.substring("mem:".length()));
|
||||
}
|
||||
String fileLocation = location.startsWith("file:")
|
||||
? location.substring("file:".length())
|
||||
: location;
|
||||
return !isRemote(fileLocation) && (location.startsWith("file:")
|
||||
|| location.startsWith("./")
|
||||
|| location.startsWith("../")
|
||||
|| location.startsWith(".\\")
|
||||
|| location.startsWith("..\\")
|
||||
|| location.startsWith("~/")
|
||||
|| location.startsWith("/")
|
||||
|| isWindowsDrive(location));
|
||||
}
|
||||
|
||||
static boolean matchesConfiguredSource(String configuredUrl, String actualUrl) {
|
||||
String configuredLocation = sourceLocation(configuredUrl);
|
||||
String actualLocation = sourceLocation(actualUrl);
|
||||
if (configuredLocation == null || actualLocation == null) {
|
||||
return false;
|
||||
}
|
||||
if (configuredLocation.startsWith("mem:") || actualLocation.startsWith("mem:")) {
|
||||
return configuredLocation.equals(actualLocation);
|
||||
}
|
||||
try {
|
||||
return localPath(configuredLocation).equals(localPath(actualLocation));
|
||||
} catch (RuntimeException exception) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private static String sourceLocation(String jdbcUrl) {
|
||||
if (jdbcUrl == null || !jdbcUrl.regionMatches(true, 0, H2_PREFIX, 0, H2_PREFIX.length())) {
|
||||
return null;
|
||||
}
|
||||
String location = jdbcUrl.substring(H2_PREFIX.length()).split(";", 2)[0];
|
||||
return location.regionMatches(true, 0, "file:", 0, "file:".length())
|
||||
? location.substring("file:".length()) : location;
|
||||
}
|
||||
|
||||
private static Path localPath(String location) {
|
||||
String expanded = location.startsWith("~/") || location.startsWith("~\\")
|
||||
? System.getProperty("user.home") + location.substring(1) : location;
|
||||
return Path.of(expanded).toAbsolutePath().normalize();
|
||||
}
|
||||
|
||||
private static boolean isRemote(String location) {
|
||||
return location.contains("://") || location.startsWith("//") || location.startsWith("\\\\");
|
||||
}
|
||||
|
||||
private static boolean isWindowsDrive(String location) {
|
||||
return location.length() >= 3
|
||||
&& Character.isLetter(location.charAt(0))
|
||||
&& location.charAt(1) == ':'
|
||||
&& (location.charAt(2) == '\\' || location.charAt(2) == '/');
|
||||
}
|
||||
|
||||
private static boolean hasUnsafeSetting(String url) {
|
||||
for (String setting : url.split(";")) {
|
||||
String normalized = setting.strip();
|
||||
if (normalized.startsWith("auto_server") || normalized.equals("file_lock=no")) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
+101
@@ -0,0 +1,101 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.manager.maintenance;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.sql.DatabaseMetaData;
|
||||
import java.sql.SQLException;
|
||||
import java.time.Duration;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import javax.sql.DataSource;
|
||||
import org.apache.hertzbeat.common.runtime.ConditionalOnNormalBusinessRuntime;
|
||||
import org.springframework.beans.factory.DisposableBean;
|
||||
import org.springframework.boot.jdbc.autoconfigure.DataSourceProperties;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/** Holds an embedded H2 connection after proving its configured local access mode is safe. */
|
||||
@Component
|
||||
@ConditionalOnNormalBusinessRuntime
|
||||
public final class EmbeddedH2SourceGuard implements MigrationSourceGuard, DisposableBean {
|
||||
|
||||
private final DataSourceProperties dataSourceProperties;
|
||||
private final DeadlineConnectionAcquirer connectionAcquirer;
|
||||
|
||||
public EmbeddedH2SourceGuard(DataSource dataSource, DataSourceProperties dataSourceProperties) {
|
||||
this.dataSourceProperties = dataSourceProperties;
|
||||
this.connectionAcquirer = new DeadlineConnectionAcquirer(dataSource);
|
||||
}
|
||||
|
||||
@Override
|
||||
public MigrationSourceLease fence(String operationId, Duration timeout) {
|
||||
requireRequest(operationId, timeout);
|
||||
String configuredUrl = dataSourceProperties.getUrl();
|
||||
if (!EmbeddedH2SourceClassifier.isSafeEmbeddedSource("H2", configuredUrl)) {
|
||||
throw MigrationMaintenanceException.sourceUnavailable();
|
||||
}
|
||||
Connection connection = connectionAcquirer.acquire(timeout);
|
||||
try {
|
||||
DatabaseMetaData metadata = connection.getMetaData();
|
||||
String productName = metadata.getDatabaseProductName();
|
||||
String actualUrl = metadata.getURL();
|
||||
if (!EmbeddedH2SourceClassifier.isSafeEmbeddedSource(productName, actualUrl)
|
||||
|| !EmbeddedH2SourceClassifier.matchesConfiguredSource(configuredUrl, actualUrl)) {
|
||||
closeRejectedConnection(connection);
|
||||
throw MigrationMaintenanceException.sourceUnavailable();
|
||||
}
|
||||
return new ConnectionSourceLease(connection);
|
||||
} catch (MigrationMaintenanceException exception) {
|
||||
throw exception;
|
||||
} catch (SQLException | RuntimeException exception) {
|
||||
closeRejectedConnection(connection);
|
||||
throw MigrationMaintenanceException.sourceUnavailable();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void destroy() {
|
||||
connectionAcquirer.close();
|
||||
}
|
||||
|
||||
private void requireRequest(String operationId, Duration timeout) {
|
||||
if (operationId == null || operationId.isBlank() || timeout == null || timeout.isNegative()) {
|
||||
throw MigrationMaintenanceException.invalidRequest();
|
||||
}
|
||||
}
|
||||
|
||||
private void closeRejectedConnection(Connection connection) {
|
||||
try {
|
||||
connection.close();
|
||||
} catch (SQLException | RuntimeException exception) {
|
||||
// The stable source failure remains primary.
|
||||
}
|
||||
}
|
||||
|
||||
private static final class ConnectionSourceLease implements MigrationSourceLease {
|
||||
|
||||
private final Connection connection;
|
||||
private final AtomicBoolean closed = new AtomicBoolean();
|
||||
|
||||
private ConnectionSourceLease(Connection connection) {
|
||||
this.connection = connection;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
if (!closed.compareAndSet(false, true)) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
connection.close();
|
||||
} catch (SQLException | RuntimeException exception) {
|
||||
closed.set(false);
|
||||
throw MigrationMaintenanceException.resumeFailure();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+17
@@ -73,6 +73,23 @@ public final class MetadataMaintenanceCoordinator {
|
||||
}
|
||||
}
|
||||
|
||||
void recover(String requestedOperationId) {
|
||||
requireOperationId(requestedOperationId);
|
||||
lock.lock();
|
||||
try {
|
||||
if (phase != MetadataMaintenancePhase.RECOVERY_REQUIRED
|
||||
|| !requestedOperationId.equals(operationId)) {
|
||||
throw MetadataMaintenanceException.operationConflict();
|
||||
}
|
||||
if (!resumeAllParticipants()) {
|
||||
throw MetadataMaintenanceException.resumeFailure();
|
||||
}
|
||||
reopen();
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
void resume(String resumedOperationId, long resumedEpoch, Object resumedToken) {
|
||||
lock.lock();
|
||||
try {
|
||||
|
||||
+33
@@ -0,0 +1,33 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.manager.maintenance;
|
||||
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
/** Fail-closed fallbacks for maintenance facts that the runtime cannot prove. */
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
public class MigrationGuardConfiguration {
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean(DeploymentSingletonAuthority.class)
|
||||
DeploymentSingletonAuthority unavailableDeploymentSingletonAuthority() {
|
||||
return (operationId, timeout) -> {
|
||||
throw MigrationMaintenanceException.deploymentAuthorityUnavailable();
|
||||
};
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean(MigrationSourceGuard.class)
|
||||
MigrationSourceGuard unavailableMigrationSourceGuard() {
|
||||
return (operationId, timeout) -> {
|
||||
throw MigrationMaintenanceException.sourceUnavailable();
|
||||
};
|
||||
}
|
||||
}
|
||||
+21
@@ -0,0 +1,21 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.manager.maintenance;
|
||||
|
||||
/** Stable safe failure categories for migration maintenance acquisition and release. */
|
||||
public enum MigrationMaintenanceErrorCode {
|
||||
MIGRATION_DEPLOYMENT_AUTHORITY_UNAVAILABLE,
|
||||
MIGRATION_SOURCE_UNAVAILABLE,
|
||||
MIGRATION_MULTI_NODE_UNSUPPORTED,
|
||||
MIGRATION_OPERATION_CONFLICT,
|
||||
MIGRATION_MAINTENANCE_TIMEOUT,
|
||||
MIGRATION_MAINTENANCE_INTERRUPTED,
|
||||
MIGRATION_MAINTENANCE_FAILURE,
|
||||
MIGRATION_RESUME_FAILURE,
|
||||
INVALID_REQUEST
|
||||
}
|
||||
+77
@@ -0,0 +1,77 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.manager.maintenance;
|
||||
|
||||
/** Secret-free migration maintenance failure. */
|
||||
public final class MigrationMaintenanceException extends RuntimeException {
|
||||
|
||||
private final MigrationMaintenanceErrorCode code;
|
||||
|
||||
private MigrationMaintenanceException(MigrationMaintenanceErrorCode code, String message) {
|
||||
super(message);
|
||||
this.code = code;
|
||||
}
|
||||
|
||||
public MigrationMaintenanceErrorCode code() {
|
||||
return code;
|
||||
}
|
||||
|
||||
public String safeMessage() {
|
||||
return getMessage();
|
||||
}
|
||||
|
||||
public static MigrationMaintenanceException deploymentAuthorityUnavailable() {
|
||||
return failure(MigrationMaintenanceErrorCode.MIGRATION_DEPLOYMENT_AUTHORITY_UNAVAILABLE,
|
||||
"Migration deployment authority is unavailable");
|
||||
}
|
||||
|
||||
public static MigrationMaintenanceException sourceUnavailable() {
|
||||
return failure(MigrationMaintenanceErrorCode.MIGRATION_SOURCE_UNAVAILABLE,
|
||||
"Migration metadata source is unavailable");
|
||||
}
|
||||
|
||||
public static MigrationMaintenanceException multiNodeUnsupported() {
|
||||
return failure(MigrationMaintenanceErrorCode.MIGRATION_MULTI_NODE_UNSUPPORTED,
|
||||
"Multi-node metadata migration is unsupported");
|
||||
}
|
||||
|
||||
static MigrationMaintenanceException operationConflict() {
|
||||
return failure(MigrationMaintenanceErrorCode.MIGRATION_OPERATION_CONFLICT,
|
||||
"Migration maintenance operation is already active");
|
||||
}
|
||||
|
||||
static MigrationMaintenanceException timeout() {
|
||||
return failure(MigrationMaintenanceErrorCode.MIGRATION_MAINTENANCE_TIMEOUT,
|
||||
"Migration maintenance acquisition timed out");
|
||||
}
|
||||
|
||||
static MigrationMaintenanceException interrupted() {
|
||||
return failure(MigrationMaintenanceErrorCode.MIGRATION_MAINTENANCE_INTERRUPTED,
|
||||
"Migration maintenance acquisition was interrupted");
|
||||
}
|
||||
|
||||
static MigrationMaintenanceException maintenanceFailure() {
|
||||
return failure(MigrationMaintenanceErrorCode.MIGRATION_MAINTENANCE_FAILURE,
|
||||
"Migration maintenance acquisition failed");
|
||||
}
|
||||
|
||||
static MigrationMaintenanceException resumeFailure() {
|
||||
return failure(MigrationMaintenanceErrorCode.MIGRATION_RESUME_FAILURE,
|
||||
"Migration maintenance release failed");
|
||||
}
|
||||
|
||||
static MigrationMaintenanceException invalidRequest() {
|
||||
return failure(MigrationMaintenanceErrorCode.INVALID_REQUEST,
|
||||
"Migration maintenance request is invalid");
|
||||
}
|
||||
|
||||
private static MigrationMaintenanceException failure(
|
||||
MigrationMaintenanceErrorCode code, String message) {
|
||||
return new MigrationMaintenanceException(code, message);
|
||||
}
|
||||
}
|
||||
+15
@@ -0,0 +1,15 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.manager.maintenance;
|
||||
|
||||
/** Owner capability for one fully acquired migration maintenance window. */
|
||||
public interface MigrationMaintenanceLease extends AutoCloseable {
|
||||
|
||||
@Override
|
||||
void close();
|
||||
}
|
||||
+16
@@ -0,0 +1,16 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.manager.maintenance;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
/** Acquires the complete process-local maintenance window required before metadata migration. */
|
||||
public interface MigrationMaintenanceOrchestrator {
|
||||
|
||||
MigrationMaintenanceLease acquire(String operationId, Duration timeout);
|
||||
}
|
||||
+16
@@ -0,0 +1,16 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.manager.maintenance;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
/** Fences a metadata source whose local access mode is safe for migration. */
|
||||
public interface MigrationSourceGuard {
|
||||
|
||||
MigrationSourceLease fence(String operationId, Duration timeout);
|
||||
}
|
||||
+15
@@ -0,0 +1,15 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.manager.maintenance;
|
||||
|
||||
/** Capability that releases one safe metadata-source fence. */
|
||||
public interface MigrationSourceLease extends AutoCloseable {
|
||||
|
||||
@Override
|
||||
void close();
|
||||
}
|
||||
+355
@@ -0,0 +1,355 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.manager.maintenance;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.eq;
|
||||
import static org.mockito.Mockito.inOrder;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import org.apache.hertzbeat.common.transaction.MetadataWriteAdmissionCoordinator;
|
||||
import org.apache.hertzbeat.common.transaction.MetadataWriteMaintenanceLease;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.InOrder;
|
||||
import org.mockito.Mockito;
|
||||
|
||||
class DefaultMigrationMaintenanceOrchestratorTest {
|
||||
|
||||
@Test
|
||||
void acquiresForwardWithOneDeadlineAndReleasesInReverse() {
|
||||
Harness harness = harness();
|
||||
AtomicLong ticker = new AtomicLong(10);
|
||||
DefaultMigrationMaintenanceOrchestrator orchestrator = new DefaultMigrationMaintenanceOrchestrator(
|
||||
harness.deploymentAuthority, harness.sourceGuard,
|
||||
harness.producerCoordinator, harness.writeCoordinator, ticker::get);
|
||||
when(harness.deploymentAuthority.acquire(eq("operation-a"), any())).thenAnswer(invocation -> {
|
||||
assertThat((Duration) invocation.getArgument(1)).isEqualTo(Duration.ofNanos(100));
|
||||
ticker.addAndGet(10);
|
||||
return harness.authorityLease;
|
||||
});
|
||||
when(harness.sourceGuard.fence(eq("operation-a"), any())).thenAnswer(invocation -> {
|
||||
assertThat((Duration) invocation.getArgument(1)).isEqualTo(Duration.ofNanos(90));
|
||||
ticker.addAndGet(20);
|
||||
return harness.sourceLease;
|
||||
});
|
||||
when(harness.producerCoordinator.quiesce(eq("operation-a"), any())).thenAnswer(invocation -> {
|
||||
assertThat((Duration) invocation.getArgument(1)).isEqualTo(Duration.ofNanos(70));
|
||||
ticker.addAndGet(30);
|
||||
return harness.producerLease;
|
||||
});
|
||||
when(harness.writeCoordinator.acquire(eq("operation-a"), any())).thenAnswer(invocation -> {
|
||||
assertThat((Duration) invocation.getArgument(1)).isEqualTo(Duration.ofNanos(40));
|
||||
return harness.writeLease;
|
||||
});
|
||||
|
||||
MigrationMaintenanceLease lease = orchestrator.acquire("operation-a", Duration.ofNanos(100));
|
||||
lease.close();
|
||||
|
||||
InOrder order = inOrder(harness.deploymentAuthority, harness.sourceGuard, harness.producerCoordinator,
|
||||
harness.writeCoordinator, harness.writeLease, harness.producerLease,
|
||||
harness.sourceLease, harness.authorityLease);
|
||||
order.verify(harness.deploymentAuthority).acquire(eq("operation-a"), any());
|
||||
order.verify(harness.sourceGuard).fence(eq("operation-a"), any());
|
||||
order.verify(harness.producerCoordinator).quiesce(eq("operation-a"), any());
|
||||
order.verify(harness.writeCoordinator).acquire(eq("operation-a"), any());
|
||||
order.verify(harness.writeLease).close();
|
||||
order.verify(harness.producerLease).resume();
|
||||
order.verify(harness.sourceLease).close();
|
||||
order.verify(harness.authorityLease).close();
|
||||
}
|
||||
|
||||
@Test
|
||||
void deploymentAuthorityUnknownOrMultiFailsBeforeSourceOrLocalPause() {
|
||||
Harness unknown = harness();
|
||||
when(unknown.deploymentAuthority.acquire(eq("operation-a"), any()))
|
||||
.thenThrow(MigrationMaintenanceException.deploymentAuthorityUnavailable());
|
||||
DefaultMigrationMaintenanceOrchestrator unknownOrchestrator = unknown.orchestrator();
|
||||
assertThatThrownBy(() -> unknownOrchestrator.acquire("operation-a", Duration.ZERO))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code())
|
||||
.isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_DEPLOYMENT_AUTHORITY_UNAVAILABLE));
|
||||
verify(unknown.sourceGuard, never()).fence(any(), any());
|
||||
verify(unknown.producerCoordinator, never()).quiesce(any(), any());
|
||||
verify(unknown.writeCoordinator, never()).acquire(any(), any());
|
||||
|
||||
Harness multi = harness();
|
||||
when(multi.deploymentAuthority.acquire(eq("operation-a"), any()))
|
||||
.thenThrow(MigrationMaintenanceException.multiNodeUnsupported());
|
||||
assertThatThrownBy(() -> multi.orchestrator().acquire("operation-a", Duration.ZERO))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code())
|
||||
.isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_MULTI_NODE_UNSUPPORTED));
|
||||
verify(multi.producerCoordinator, never()).quiesce(any(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void secondAndThirdStepFailuresCleanUpInReverseWithoutReplacingPrimary() {
|
||||
Harness second = harness();
|
||||
when(second.sourceGuard.fence(eq("operation-a"), any())).thenReturn(second.sourceLease);
|
||||
when(second.producerCoordinator.quiesce(eq("operation-a"), any()))
|
||||
.thenThrow(MetadataMaintenanceException.quiesceTimeout());
|
||||
Mockito.doThrow(new IllegalStateException("private-cleanup")).when(second.sourceLease).close();
|
||||
assertThatThrownBy(() -> second.orchestrator().acquire("operation-a", Duration.ZERO))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception -> {
|
||||
assertThat(exception.code()).isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_MAINTENANCE_TIMEOUT);
|
||||
assertThat(exception.getSuppressed()).hasSize(1);
|
||||
assertThat(exception.getMessage()).doesNotContain("private");
|
||||
});
|
||||
|
||||
Harness third = harness();
|
||||
when(third.sourceGuard.fence(eq("operation-a"), any())).thenReturn(third.sourceLease);
|
||||
when(third.producerCoordinator.quiesce(eq("operation-a"), any())).thenReturn(third.producerLease);
|
||||
when(third.writeCoordinator.acquire(eq("operation-a"), any()))
|
||||
.thenThrow(new IllegalStateException("private-write"));
|
||||
assertThatThrownBy(() -> third.orchestrator().acquire("operation-a", Duration.ZERO))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code())
|
||||
.isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_MAINTENANCE_FAILURE));
|
||||
InOrder cleanup = inOrder(third.producerLease, third.sourceLease, third.authorityLease);
|
||||
cleanup.verify(third.producerLease).resume();
|
||||
cleanup.verify(third.sourceLease).close();
|
||||
cleanup.verify(third.authorityLease).close();
|
||||
}
|
||||
|
||||
@Test
|
||||
void failedAcquisitionRecoveryRetainsFenceAndOwnershipUntilSameOperationRetries() {
|
||||
Harness harness = harness();
|
||||
when(harness.sourceGuard.fence(eq("operation-a"), any()))
|
||||
.thenReturn(harness.sourceLease)
|
||||
.thenThrow(MigrationMaintenanceException.sourceUnavailable());
|
||||
when(harness.producerCoordinator.quiesce(eq("operation-a"), any())).thenReturn(harness.producerLease);
|
||||
when(harness.writeCoordinator.acquire(eq("operation-a"), any()))
|
||||
.thenThrow(new IllegalStateException("private-write"));
|
||||
Mockito.doThrow(new IllegalStateException("private-resume"))
|
||||
.doNothing().when(harness.producerLease).resume();
|
||||
DefaultMigrationMaintenanceOrchestrator orchestrator = harness.orchestrator();
|
||||
|
||||
assertThatThrownBy(() -> orchestrator.acquire("operation-a", Duration.ZERO))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception -> {
|
||||
assertThat(exception.code())
|
||||
.isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_MAINTENANCE_FAILURE);
|
||||
assertThat(exception.getSuppressed()).hasSize(1);
|
||||
});
|
||||
verify(harness.sourceLease, never()).close();
|
||||
verify(harness.authorityLease, never()).close();
|
||||
assertThatThrownBy(() -> orchestrator.acquire("operation-b", Duration.ZERO))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code())
|
||||
.isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_OPERATION_CONFLICT));
|
||||
|
||||
assertThatThrownBy(() -> orchestrator.acquire("operation-a", Duration.ZERO))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code())
|
||||
.isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_SOURCE_UNAVAILABLE));
|
||||
verify(harness.producerLease, Mockito.times(2)).resume();
|
||||
verify(harness.sourceLease).close();
|
||||
verify(harness.authorityLease, Mockito.times(2)).close();
|
||||
}
|
||||
|
||||
@Test
|
||||
void producerRecoveryWithoutLeaseRetainsFenceUntilSameOperationRecoversCoordinator() {
|
||||
Harness harness = harness();
|
||||
when(harness.sourceGuard.fence(eq("operation-a"), any()))
|
||||
.thenReturn(harness.sourceLease)
|
||||
.thenThrow(MigrationMaintenanceException.sourceUnavailable());
|
||||
when(harness.producerCoordinator.quiesce(eq("operation-a"), any()))
|
||||
.thenThrow(MetadataMaintenanceException.participantFailure());
|
||||
AtomicReference<MetadataMaintenanceSnapshot> producerState = new AtomicReference<>(
|
||||
new MetadataMaintenanceSnapshot(MetadataMaintenancePhase.RECOVERY_REQUIRED, "operation-a", 1));
|
||||
when(harness.producerCoordinator.snapshot()).thenAnswer(invocation -> producerState.get());
|
||||
Mockito.doAnswer(invocation -> {
|
||||
producerState.set(new MetadataMaintenanceSnapshot(MetadataMaintenancePhase.RUNNING, null, 1));
|
||||
return null;
|
||||
}).when(harness.producerCoordinator).recover("operation-a");
|
||||
DefaultMigrationMaintenanceOrchestrator orchestrator = harness.orchestrator();
|
||||
|
||||
assertThatThrownBy(() -> orchestrator.acquire("operation-a", Duration.ofSeconds(1)))
|
||||
.isInstanceOf(MigrationMaintenanceException.class);
|
||||
verify(harness.sourceLease, never()).close();
|
||||
assertThatThrownBy(() -> orchestrator.acquire("operation-b", Duration.ZERO))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code()).isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_OPERATION_CONFLICT));
|
||||
|
||||
assertThatThrownBy(() -> orchestrator.acquire("operation-a", Duration.ofSeconds(1)))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code())
|
||||
.isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_SOURCE_UNAVAILABLE));
|
||||
verify(harness.producerCoordinator).recover("operation-a");
|
||||
verify(harness.sourceLease).close();
|
||||
verify(harness.authorityLease, Mockito.times(2)).close();
|
||||
}
|
||||
|
||||
@Test
|
||||
void resumeFailureRetainsTopologyAndSameOwnerCanRetry() {
|
||||
Harness harness = harness();
|
||||
when(harness.sourceGuard.fence(eq("operation-a"), any())).thenReturn(harness.sourceLease);
|
||||
when(harness.producerCoordinator.quiesce(eq("operation-a"), any())).thenReturn(harness.producerLease);
|
||||
when(harness.writeCoordinator.acquire(eq("operation-a"), any())).thenReturn(harness.writeLease);
|
||||
Mockito.doThrow(new IllegalStateException()).doNothing().when(harness.producerLease).resume();
|
||||
MigrationMaintenanceLease lease = harness.orchestrator().acquire("operation-a", Duration.ZERO);
|
||||
|
||||
assertThatThrownBy(lease::close)
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code()).isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_RESUME_FAILURE));
|
||||
verify(harness.sourceLease, never()).close();
|
||||
verify(harness.authorityLease, never()).close();
|
||||
lease.close();
|
||||
verify(harness.writeLease).close();
|
||||
verify(harness.producerLease, Mockito.times(2)).resume();
|
||||
verify(harness.sourceLease).close();
|
||||
verify(harness.authorityLease).close();
|
||||
}
|
||||
|
||||
@Test
|
||||
void concurrentAcquisitionHasOnlyOneOwnerCapability() throws Exception {
|
||||
Harness harness = harness();
|
||||
CountDownLatch entered = new CountDownLatch(1);
|
||||
CountDownLatch release = new CountDownLatch(1);
|
||||
when(harness.sourceGuard.fence(eq("operation-a"), any())).thenAnswer(invocation -> {
|
||||
entered.countDown();
|
||||
release.await();
|
||||
return harness.sourceLease;
|
||||
});
|
||||
when(harness.producerCoordinator.quiesce(eq("operation-a"), any())).thenReturn(harness.producerLease);
|
||||
when(harness.writeCoordinator.acquire(eq("operation-a"), any())).thenReturn(harness.writeLease);
|
||||
DefaultMigrationMaintenanceOrchestrator orchestrator = harness.orchestrator();
|
||||
AtomicReference<MigrationMaintenanceLease> owner = new AtomicReference<>();
|
||||
Thread thread = Thread.ofPlatform().start(() ->
|
||||
owner.set(orchestrator.acquire("operation-a", Duration.ofSeconds(30))));
|
||||
assertThat(entered.await(1, TimeUnit.SECONDS)).isTrue();
|
||||
|
||||
assertThatThrownBy(() -> orchestrator.acquire("operation-b", Duration.ZERO))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code()).isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_OPERATION_CONFLICT));
|
||||
assertThatThrownBy(() -> orchestrator.acquire("operation-a", Duration.ZERO))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code()).isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_OPERATION_CONFLICT));
|
||||
release.countDown();
|
||||
thread.join(1_000);
|
||||
owner.get().close();
|
||||
}
|
||||
|
||||
@Test
|
||||
void sharedAuthorityNotSeparateEmbeddedSourcesDeterminesSingletonOwnership() {
|
||||
Harness first = harness();
|
||||
Harness second = harness();
|
||||
FakeSingletonAuthority authority = new FakeSingletonAuthority();
|
||||
when(first.sourceGuard.fence(eq("operation-a"), any())).thenReturn(first.sourceLease);
|
||||
when(first.producerCoordinator.quiesce(eq("operation-a"), any())).thenReturn(first.producerLease);
|
||||
when(first.writeCoordinator.acquire(eq("operation-a"), any())).thenReturn(first.writeLease);
|
||||
when(second.sourceGuard.fence(eq("operation-b"), any())).thenReturn(second.sourceLease);
|
||||
when(second.producerCoordinator.quiesce(eq("operation-b"), any())).thenReturn(second.producerLease);
|
||||
when(second.writeCoordinator.acquire(eq("operation-b"), any())).thenReturn(second.writeLease);
|
||||
DefaultMigrationMaintenanceOrchestrator firstOrchestrator = new DefaultMigrationMaintenanceOrchestrator(
|
||||
authority, first.sourceGuard, first.producerCoordinator, first.writeCoordinator);
|
||||
DefaultMigrationMaintenanceOrchestrator secondOrchestrator = new DefaultMigrationMaintenanceOrchestrator(
|
||||
authority, second.sourceGuard, second.producerCoordinator, second.writeCoordinator);
|
||||
|
||||
MigrationMaintenanceLease firstLease = firstOrchestrator.acquire("operation-a", Duration.ZERO);
|
||||
assertThatThrownBy(() -> secondOrchestrator.acquire("operation-b", Duration.ZERO))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code()).isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_OPERATION_CONFLICT));
|
||||
verify(second.sourceGuard, never()).fence(any(), any());
|
||||
firstLease.close();
|
||||
secondOrchestrator.acquire("operation-b", Duration.ZERO).close();
|
||||
}
|
||||
|
||||
@Test
|
||||
void validatesNegativeAndOverflowButAllowsZeroAndNewInstanceStartsIdle() {
|
||||
Harness first = harness();
|
||||
when(first.sourceGuard.fence(eq("operation-a"), any())).thenReturn(first.sourceLease);
|
||||
when(first.producerCoordinator.quiesce(eq("operation-a"), any())).thenReturn(first.producerLease);
|
||||
when(first.writeCoordinator.acquire(eq("operation-a"), any())).thenReturn(first.writeLease);
|
||||
MigrationMaintenanceLease lease = first.orchestrator().acquire("operation-a", Duration.ZERO);
|
||||
assertThatThrownBy(() -> first.orchestrator().acquire("operation-b", Duration.ofSeconds(-1)))
|
||||
.isInstanceOf(MigrationMaintenanceException.class);
|
||||
assertThatThrownBy(() -> first.orchestrator().acquire("operation-b", Duration.ofSeconds(Long.MAX_VALUE)))
|
||||
.isInstanceOf(MigrationMaintenanceException.class);
|
||||
lease.close();
|
||||
|
||||
Harness restarted = harness();
|
||||
when(restarted.sourceGuard.fence(eq("operation-b"), any())).thenReturn(restarted.sourceLease);
|
||||
when(restarted.producerCoordinator.quiesce(eq("operation-b"), any())).thenReturn(restarted.producerLease);
|
||||
when(restarted.writeCoordinator.acquire(eq("operation-b"), any())).thenReturn(restarted.writeLease);
|
||||
restarted.orchestrator().acquire("operation-b", Duration.ZERO).close();
|
||||
}
|
||||
|
||||
@Test
|
||||
void interruptClassificationKeepsInterruptBit() {
|
||||
Harness harness = harness();
|
||||
when(harness.deploymentAuthority.acquire(eq("operation-a"), any())).thenAnswer(invocation -> {
|
||||
Thread.currentThread().interrupt();
|
||||
throw MigrationMaintenanceException.interrupted();
|
||||
});
|
||||
|
||||
assertThatThrownBy(() -> harness.orchestrator().acquire("operation-a", Duration.ZERO))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code())
|
||||
.isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_MAINTENANCE_INTERRUPTED));
|
||||
assertThat(Thread.interrupted()).isTrue();
|
||||
}
|
||||
|
||||
private static Harness harness() {
|
||||
Harness harness = new Harness(
|
||||
Mockito.mock(DeploymentSingletonAuthority.class),
|
||||
Mockito.mock(DeploymentSingletonLease.class),
|
||||
Mockito.mock(MigrationSourceGuard.class),
|
||||
Mockito.mock(MigrationSourceLease.class),
|
||||
Mockito.mock(MetadataMaintenanceCoordinator.class),
|
||||
Mockito.mock(MetadataMaintenanceLease.class),
|
||||
Mockito.mock(MetadataWriteAdmissionCoordinator.class),
|
||||
Mockito.mock(MetadataWriteMaintenanceLease.class));
|
||||
when(harness.deploymentAuthority.acquire(any(), any())).thenReturn(harness.authorityLease);
|
||||
when(harness.producerCoordinator.snapshot()).thenReturn(
|
||||
new MetadataMaintenanceSnapshot(MetadataMaintenancePhase.RUNNING, null, 0));
|
||||
return harness;
|
||||
}
|
||||
|
||||
private record Harness(
|
||||
DeploymentSingletonAuthority deploymentAuthority,
|
||||
DeploymentSingletonLease authorityLease,
|
||||
MigrationSourceGuard sourceGuard,
|
||||
MigrationSourceLease sourceLease,
|
||||
MetadataMaintenanceCoordinator producerCoordinator,
|
||||
MetadataMaintenanceLease producerLease,
|
||||
MetadataWriteAdmissionCoordinator writeCoordinator,
|
||||
MetadataWriteMaintenanceLease writeLease) {
|
||||
|
||||
DefaultMigrationMaintenanceOrchestrator orchestrator() {
|
||||
return new DefaultMigrationMaintenanceOrchestrator(
|
||||
deploymentAuthority, sourceGuard, producerCoordinator, writeCoordinator);
|
||||
}
|
||||
}
|
||||
|
||||
private static final class FakeSingletonAuthority implements DeploymentSingletonAuthority {
|
||||
|
||||
private boolean owned;
|
||||
|
||||
@Override
|
||||
public synchronized DeploymentSingletonLease acquire(String operationId, Duration timeout) {
|
||||
if (owned) {
|
||||
throw MigrationMaintenanceException.operationConflict();
|
||||
}
|
||||
owned = true;
|
||||
return this::release;
|
||||
}
|
||||
|
||||
private synchronized void release() {
|
||||
owned = false;
|
||||
}
|
||||
}
|
||||
}
|
||||
+48
@@ -0,0 +1,48 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.manager.maintenance;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
class EmbeddedH2SourceClassifierTest {
|
||||
|
||||
@Test
|
||||
void acceptsOnlyExplicitEmbeddedMemoryAndLocalFileUrls() {
|
||||
assertThat(EmbeddedH2SourceClassifier.isSafeEmbeddedSource("H2", "jdbc:h2:mem:manager")).isTrue();
|
||||
assertThat(EmbeddedH2SourceClassifier.isSafeEmbeddedSource("H2", "jdbc:h2:file:/var/lib/manager")).isTrue();
|
||||
assertThat(EmbeddedH2SourceClassifier.isSafeEmbeddedSource("H2", "jdbc:h2:./data/manager")).isTrue();
|
||||
assertThat(EmbeddedH2SourceClassifier.isSafeEmbeddedSource("H2", "jdbc:h2:.\\data\\manager")).isTrue();
|
||||
assertThat(EmbeddedH2SourceClassifier.isSafeEmbeddedSource("H2", "jdbc:h2:..\\data\\manager")).isTrue();
|
||||
assertThat(EmbeddedH2SourceClassifier.isSafeEmbeddedSource("H2", "jdbc:h2:C:\\data\\manager")).isTrue();
|
||||
assertThat(EmbeddedH2SourceClassifier.isSafeEmbeddedSource("H2", "jdbc:h2:tcp://db/manager")).isFalse();
|
||||
assertThat(EmbeddedH2SourceClassifier.isSafeEmbeddedSource("H2", "jdbc:h2:ssl://db/manager")).isFalse();
|
||||
assertThat(EmbeddedH2SourceClassifier.isSafeEmbeddedSource("H2", "jdbc:h2:file:tcp://db/manager")).isFalse();
|
||||
assertThat(EmbeddedH2SourceClassifier.isSafeEmbeddedSource("H2", "jdbc:h2:file:////db/manager")).isFalse();
|
||||
assertThat(EmbeddedH2SourceClassifier.isSafeEmbeddedSource("H2", "jdbc:h2:\\\\db\\manager")).isFalse();
|
||||
assertThat(EmbeddedH2SourceClassifier.isSafeEmbeddedSource(
|
||||
"H2", "jdbc:h2:file:/var/lib/manager;AUTO_SERVER=TRUE")).isFalse();
|
||||
assertThat(EmbeddedH2SourceClassifier.isSafeEmbeddedSource(
|
||||
"H2", "jdbc:h2:file:/var/lib/manager;FILE_LOCK=NO")).isFalse();
|
||||
assertThat(EmbeddedH2SourceClassifier.isSafeEmbeddedSource("PostgreSQL", "jdbc:h2:mem:manager")).isFalse();
|
||||
assertThat(EmbeddedH2SourceClassifier.isSafeEmbeddedSource("H2", "jdbc:unknown:manager")).isFalse();
|
||||
}
|
||||
|
||||
@Test
|
||||
void matchesConfiguredSourceToActualMetadataLocation() {
|
||||
assertThat(EmbeddedH2SourceClassifier.matchesConfiguredSource(
|
||||
"jdbc:h2:mem:manager;MODE=MYSQL", "jdbc:h2:mem:manager")).isTrue();
|
||||
assertThat(EmbeddedH2SourceClassifier.matchesConfiguredSource(
|
||||
"jdbc:h2:./data/manager;MODE=MYSQL", "jdbc:h2:file:./data/manager")).isTrue();
|
||||
assertThat(EmbeddedH2SourceClassifier.matchesConfiguredSource(
|
||||
"jdbc:h2:mem:manager", "jdbc:h2:mem:other")).isFalse();
|
||||
assertThat(EmbeddedH2SourceClassifier.matchesConfiguredSource(
|
||||
"jdbc:h2:file:/safe/manager", "jdbc:h2:file:/other/manager")).isFalse();
|
||||
}
|
||||
}
|
||||
+232
@@ -0,0 +1,232 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.manager.maintenance;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
import java.nio.file.Path;
|
||||
import java.sql.Connection;
|
||||
import java.sql.DatabaseMetaData;
|
||||
import java.sql.DriverManager;
|
||||
import java.time.Duration;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import javax.sql.DataSource;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.io.TempDir;
|
||||
import org.mockito.Mockito;
|
||||
import org.springframework.boot.jdbc.autoconfigure.DataSourceProperties;
|
||||
import org.springframework.jdbc.datasource.DriverManagerDataSource;
|
||||
|
||||
class EmbeddedH2SourceGuardTest {
|
||||
|
||||
@Test
|
||||
void holdsSafeSourceConnectionUntilLeaseClosesExactlyOnce() throws Exception {
|
||||
DataSource dataSource = Mockito.mock(DataSource.class);
|
||||
Connection connection = Mockito.mock(Connection.class);
|
||||
DatabaseMetaData metadata = Mockito.mock(DatabaseMetaData.class);
|
||||
when(dataSource.getConnection()).thenReturn(connection);
|
||||
when(connection.getMetaData()).thenReturn(metadata);
|
||||
when(metadata.getDatabaseProductName()).thenReturn("H2");
|
||||
when(metadata.getURL()).thenReturn("jdbc:h2:mem:manager");
|
||||
EmbeddedH2SourceGuard guard = guard(dataSource, "jdbc:h2:mem:manager");
|
||||
|
||||
MigrationSourceLease lease = guard.fence("operation-a", Duration.ofSeconds(1));
|
||||
|
||||
verify(connection, never()).close();
|
||||
lease.close();
|
||||
lease.close();
|
||||
verify(connection).close();
|
||||
guard.destroy();
|
||||
}
|
||||
|
||||
@Test
|
||||
void rejectsAmbiguousSourceAndClosesConnectionWithoutDetails() throws Exception {
|
||||
DataSource dataSource = Mockito.mock(DataSource.class);
|
||||
Connection connection = Mockito.mock(Connection.class);
|
||||
DatabaseMetaData metadata = Mockito.mock(DatabaseMetaData.class);
|
||||
when(dataSource.getConnection()).thenReturn(connection);
|
||||
when(connection.getMetaData()).thenReturn(metadata);
|
||||
when(metadata.getDatabaseProductName()).thenReturn("H2");
|
||||
when(metadata.getURL()).thenReturn("jdbc:h2:tcp://private-host/secret");
|
||||
EmbeddedH2SourceGuard guard = guard(dataSource, "jdbc:h2:mem:manager");
|
||||
|
||||
assertThatThrownBy(() -> guard.fence("operation-a", Duration.ofSeconds(1)))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception -> {
|
||||
assertThat(exception.code())
|
||||
.isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_SOURCE_UNAVAILABLE);
|
||||
assertThat(exception.safeMessage()).doesNotContain("private-host").doesNotContain("secret");
|
||||
assertThat(exception.getCause()).isNull();
|
||||
});
|
||||
verify(connection).close();
|
||||
guard.destroy();
|
||||
}
|
||||
|
||||
@Test
|
||||
void rejectsRawAutoServerAndFileLockSettingsThatMetadataRemoves(@TempDir Path directory) throws Exception {
|
||||
String baseUrl = "jdbc:h2:file:" + directory.resolve("manager");
|
||||
EmbeddedH2SourceGuard safeGuard = guard(new DriverManagerDataSource(baseUrl), baseUrl);
|
||||
safeGuard.fence("safe-source", Duration.ofSeconds(1)).close();
|
||||
safeGuard.destroy();
|
||||
for (String setting : new String[] {";AUTO_SERVER=TRUE", ";FILE_LOCK=NO"}) {
|
||||
String configuredUrl = baseUrl + setting;
|
||||
try (Connection connection = DriverManager.getConnection(configuredUrl)) {
|
||||
assertThat(connection.getMetaData().getURL()).doesNotContain(setting.substring(1));
|
||||
}
|
||||
DriverManagerDataSource dataSource = new DriverManagerDataSource(configuredUrl);
|
||||
assertThatThrownBy(() -> guard(dataSource, configuredUrl)
|
||||
.fence("operation-a", Duration.ofSeconds(1)))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code())
|
||||
.isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_SOURCE_UNAVAILABLE));
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void timeoutAbandonsBlockingConnectionAndClosesLateResult() throws Exception {
|
||||
assertTimedOutConnectionClosed(Duration.ZERO);
|
||||
assertTimedOutConnectionClosed(Duration.ofMillis(20));
|
||||
}
|
||||
|
||||
@Test
|
||||
void rejectsOverflowDurationWithoutStartingConnectionWork() throws Exception {
|
||||
DataSource dataSource = Mockito.mock(DataSource.class);
|
||||
EmbeddedH2SourceGuard guard = guard(dataSource, "jdbc:h2:mem:manager");
|
||||
|
||||
assertThatThrownBy(() -> guard.fence("operation-a", Duration.ofSeconds(Long.MAX_VALUE)))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code()).isEqualTo(MigrationMaintenanceErrorCode.INVALID_REQUEST));
|
||||
verify(dataSource, never()).getConnection();
|
||||
guard.destroy();
|
||||
}
|
||||
|
||||
@Test
|
||||
void destroyRejectsNewAcquisitionWithoutStartingConnectionWork() throws Exception {
|
||||
DataSource dataSource = Mockito.mock(DataSource.class);
|
||||
EmbeddedH2SourceGuard guard = guard(dataSource, "jdbc:h2:mem:manager");
|
||||
guard.destroy();
|
||||
|
||||
assertThatThrownBy(() -> guard.fence("operation-a", Duration.ofSeconds(1)))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code()).isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_SOURCE_UNAVAILABLE));
|
||||
verify(dataSource, never()).getConnection();
|
||||
}
|
||||
|
||||
private void assertTimedOutConnectionClosed(Duration timeout) throws Exception {
|
||||
DataSource dataSource = Mockito.mock(DataSource.class);
|
||||
Connection connection = Mockito.mock(Connection.class);
|
||||
CountDownLatch entered = new CountDownLatch(1);
|
||||
CountDownLatch release = new CountDownLatch(1);
|
||||
CountDownLatch closed = new CountDownLatch(1);
|
||||
when(dataSource.getConnection()).thenAnswer(invocation -> {
|
||||
entered.countDown();
|
||||
release.await();
|
||||
return connection;
|
||||
});
|
||||
Mockito.doAnswer(invocation -> {
|
||||
closed.countDown();
|
||||
return null;
|
||||
}).when(connection).close();
|
||||
EmbeddedH2SourceGuard guard = guard(dataSource, "jdbc:h2:mem:manager");
|
||||
|
||||
assertThatThrownBy(() -> guard.fence("operation-a", timeout))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code())
|
||||
.isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_MAINTENANCE_TIMEOUT));
|
||||
assertThat(entered.await(1, TimeUnit.SECONDS)).isTrue();
|
||||
release.countDown();
|
||||
assertThat(closed.await(1, TimeUnit.SECONDS)).isTrue();
|
||||
guard.destroy();
|
||||
}
|
||||
|
||||
@Test
|
||||
void interruptAbandonsBlockingConnectionAndPreservesInterrupt() throws Exception {
|
||||
DataSource dataSource = Mockito.mock(DataSource.class);
|
||||
Connection connection = Mockito.mock(Connection.class);
|
||||
CountDownLatch entered = new CountDownLatch(1);
|
||||
CountDownLatch release = new CountDownLatch(1);
|
||||
CountDownLatch closed = new CountDownLatch(1);
|
||||
when(dataSource.getConnection()).thenAnswer(invocation -> {
|
||||
entered.countDown();
|
||||
release.await();
|
||||
return connection;
|
||||
});
|
||||
Mockito.doAnswer(invocation -> {
|
||||
closed.countDown();
|
||||
return null;
|
||||
}).when(connection).close();
|
||||
AtomicReference<MigrationMaintenanceErrorCode> code = new AtomicReference<>();
|
||||
AtomicBoolean interrupted = new AtomicBoolean();
|
||||
EmbeddedH2SourceGuard guard = guard(dataSource, "jdbc:h2:mem:manager");
|
||||
Thread caller = Thread.ofPlatform().start(() -> {
|
||||
try {
|
||||
guard.fence("operation-a", Duration.ofSeconds(30));
|
||||
} catch (MigrationMaintenanceException exception) {
|
||||
code.set(exception.code());
|
||||
interrupted.set(Thread.currentThread().isInterrupted());
|
||||
}
|
||||
});
|
||||
assertThat(entered.await(1, TimeUnit.SECONDS)).isTrue();
|
||||
|
||||
caller.interrupt();
|
||||
caller.join(1_000);
|
||||
assertThat(code.get()).isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_MAINTENANCE_INTERRUPTED);
|
||||
assertThat(interrupted).isTrue();
|
||||
release.countDown();
|
||||
assertThat(closed.await(1, TimeUnit.SECONDS)).isTrue();
|
||||
guard.destroy();
|
||||
}
|
||||
|
||||
@Test
|
||||
void oneStuckDriverCallBoundsAllLaterAcquisitionWork() throws Exception {
|
||||
DataSource dataSource = Mockito.mock(DataSource.class);
|
||||
Connection connection = Mockito.mock(Connection.class);
|
||||
AtomicInteger calls = new AtomicInteger();
|
||||
CountDownLatch entered = new CountDownLatch(1);
|
||||
CountDownLatch release = new CountDownLatch(1);
|
||||
CountDownLatch closed = new CountDownLatch(1);
|
||||
when(dataSource.getConnection()).thenAnswer(invocation -> {
|
||||
calls.incrementAndGet();
|
||||
entered.countDown();
|
||||
release.await();
|
||||
return connection;
|
||||
});
|
||||
Mockito.doAnswer(invocation -> {
|
||||
closed.countDown();
|
||||
return null;
|
||||
}).when(connection).close();
|
||||
EmbeddedH2SourceGuard guard = guard(dataSource, "jdbc:h2:mem:manager");
|
||||
|
||||
assertThatThrownBy(() -> guard.fence("operation-a", Duration.ZERO))
|
||||
.isInstanceOf(MigrationMaintenanceException.class);
|
||||
assertThat(entered.await(1, TimeUnit.SECONDS)).isTrue();
|
||||
for (int index = 0; index < 20; index++) {
|
||||
assertThatThrownBy(() -> guard.fence("operation-a", Duration.ZERO))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code())
|
||||
.isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_MAINTENANCE_TIMEOUT));
|
||||
}
|
||||
assertThat(calls).hasValue(1);
|
||||
release.countDown();
|
||||
assertThat(closed.await(1, TimeUnit.SECONDS)).isTrue();
|
||||
guard.destroy();
|
||||
}
|
||||
|
||||
private static EmbeddedH2SourceGuard guard(DataSource dataSource, String configuredUrl) {
|
||||
DataSourceProperties properties = new DataSourceProperties();
|
||||
properties.setUrl(configuredUrl);
|
||||
return new EmbeddedH2SourceGuard(dataSource, properties);
|
||||
}
|
||||
}
|
||||
+12
-2
@@ -15,6 +15,7 @@ import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import org.junit.jupiter.api.Test;
|
||||
@@ -199,7 +200,8 @@ class MetadataMaintenanceCoordinatorTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
void failedParticipantRecoveryCannotBeReportedAsRunning() {
|
||||
void failedParticipantRecoveryRequiresSameOperationAndCanBeRetriedExplicitly() {
|
||||
AtomicBoolean resumeFails = new AtomicBoolean(true);
|
||||
MetadataMaintenanceParticipant participant = new MetadataMaintenanceParticipant() {
|
||||
@Override
|
||||
public String participantId() {
|
||||
@@ -213,7 +215,9 @@ class MetadataMaintenanceCoordinatorTest {
|
||||
|
||||
@Override
|
||||
public void resume() {
|
||||
throw MetadataMaintenanceException.resumeFailure();
|
||||
if (resumeFails.get()) {
|
||||
throw MetadataMaintenanceException.resumeFailure();
|
||||
}
|
||||
}
|
||||
};
|
||||
MetadataMaintenanceCoordinator coordinator = new MetadataMaintenanceCoordinator(List.of(participant));
|
||||
@@ -223,6 +227,12 @@ class MetadataMaintenanceCoordinatorTest {
|
||||
assertThat(exception.code()).isEqualTo(MetadataMaintenanceErrorCode.QUIESCE_TIMEOUT));
|
||||
|
||||
assertThat(coordinator.snapshot().phase()).isEqualTo(MetadataMaintenancePhase.RECOVERY_REQUIRED);
|
||||
assertThatThrownBy(() -> coordinator.recover("operation-b"))
|
||||
.isInstanceOfSatisfying(MetadataMaintenanceException.class, exception ->
|
||||
assertThat(exception.code()).isEqualTo(MetadataMaintenanceErrorCode.OPERATION_CONFLICT));
|
||||
resumeFails.set(false);
|
||||
coordinator.recover("operation-a");
|
||||
assertThat(coordinator.snapshot().phase()).isEqualTo(MetadataMaintenancePhase.RUNNING);
|
||||
}
|
||||
|
||||
@Test
|
||||
|
||||
+19
@@ -8,8 +8,10 @@
|
||||
package org.apache.hertzbeat.startup;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
import java.lang.reflect.Method;
|
||||
import java.time.Duration;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import org.apache.hertzbeat.common.transaction.MetadataWriteAdmissionAdvisor;
|
||||
@@ -21,6 +23,12 @@ import org.apache.hertzbeat.manager.maintenance.MetadataMaintenanceCoordinator;
|
||||
import org.apache.hertzbeat.manager.maintenance.MetadataMaintenanceParticipant;
|
||||
import org.apache.hertzbeat.manager.maintenance.AlertMetadataMaintenanceParticipant;
|
||||
import org.apache.hertzbeat.manager.maintenance.CollectorLifecycleMaintenanceParticipant;
|
||||
import org.apache.hertzbeat.manager.maintenance.DeploymentSingletonAuthority;
|
||||
import org.apache.hertzbeat.manager.maintenance.EmbeddedH2SourceGuard;
|
||||
import org.apache.hertzbeat.manager.maintenance.MigrationMaintenanceOrchestrator;
|
||||
import org.apache.hertzbeat.manager.maintenance.MigrationMaintenanceErrorCode;
|
||||
import org.apache.hertzbeat.manager.maintenance.MigrationMaintenanceException;
|
||||
import org.apache.hertzbeat.manager.maintenance.MigrationSourceGuard;
|
||||
import org.apache.hertzbeat.manager.setup.runtime.SetupRuntimeTransition;
|
||||
import org.apache.hertzbeat.warehouse.store.DataStorageDispatch;
|
||||
import org.apache.hertzbeat.warehouse.store.metadata.JdbcMonitorStatusMetadataWriter;
|
||||
@@ -52,6 +60,17 @@ class MetadataWriteAdmissionStartupContextTest {
|
||||
@Test
|
||||
void startupHasOneAdmissionBoundaryAndOneTransactionSource() throws Exception {
|
||||
assertThat(context.getBeansOfType(MetadataMaintenanceCoordinator.class)).hasSize(1);
|
||||
assertThat(context.getBeansOfType(MigrationMaintenanceOrchestrator.class)).hasSize(1);
|
||||
assertThat(context.getBeansOfType(MigrationSourceGuard.class))
|
||||
.containsOnlyKeys("embeddedH2SourceGuard");
|
||||
assertThat(context.getBean(MigrationSourceGuard.class)).isInstanceOf(EmbeddedH2SourceGuard.class);
|
||||
assertThat(context.getBeansOfType(DeploymentSingletonAuthority.class))
|
||||
.containsOnlyKeys("unavailableDeploymentSingletonAuthority");
|
||||
assertThatThrownBy(() -> context.getBean(MigrationMaintenanceOrchestrator.class)
|
||||
.acquire("normal-proof", Duration.ZERO))
|
||||
.isInstanceOfSatisfying(MigrationMaintenanceException.class, exception ->
|
||||
assertThat(exception.code())
|
||||
.isEqualTo(MigrationMaintenanceErrorCode.MIGRATION_DEPLOYMENT_AUTHORITY_UNAVAILABLE));
|
||||
List<MetadataMaintenanceParticipant> participants = context.getBeanProvider(
|
||||
MetadataMaintenanceParticipant.class).orderedStream().toList();
|
||||
assertThat(participants)
|
||||
|
||||
+13
@@ -30,10 +30,14 @@ import java.net.http.HttpRequest;
|
||||
import java.net.http.HttpResponse;
|
||||
import java.nio.file.Path;
|
||||
import java.time.Clock;
|
||||
import java.time.Duration;
|
||||
import org.apache.hertzbeat.common.runtime.BusinessRuntimeGate;
|
||||
import org.apache.hertzbeat.common.runtime.RuntimeMode;
|
||||
import org.apache.hertzbeat.common.util.JsonUtil;
|
||||
import org.apache.hertzbeat.manager.setup.api.SetupApiContract.SetupPhase;
|
||||
import org.apache.hertzbeat.manager.maintenance.MigrationMaintenanceErrorCode;
|
||||
import org.apache.hertzbeat.manager.maintenance.MigrationMaintenanceException;
|
||||
import org.apache.hertzbeat.manager.maintenance.MigrationMaintenanceOrchestrator;
|
||||
import org.apache.hertzbeat.manager.setup.api.SetupApiContract.SetupErrorCode;
|
||||
import org.apache.hertzbeat.manager.setup.api.SetupApiContract.SetupOperationState;
|
||||
import org.apache.hertzbeat.manager.setup.runtime.SetupRuntimeTransition;
|
||||
@@ -167,6 +171,15 @@ class StartupRuntimeBoundaryContextTest {
|
||||
assertTrue(context.containsBeanDefinition("periodicAlertRuleScheduler"));
|
||||
assertTrue(context.containsBeanDefinition("manageServer"));
|
||||
assertFalse(context.containsBeanDefinition("collectorLifecycleMaintenanceParticipant"));
|
||||
MigrationMaintenanceOrchestrator orchestrator =
|
||||
context.getBean(MigrationMaintenanceOrchestrator.class);
|
||||
try {
|
||||
orchestrator.acquire("gated-proof", Duration.ZERO);
|
||||
org.junit.jupiter.api.Assertions.fail("Gated runtime must fail closed");
|
||||
} catch (MigrationMaintenanceException exception) {
|
||||
assertEquals(MigrationMaintenanceErrorCode.MIGRATION_DEPLOYMENT_AUTHORITY_UNAVAILABLE,
|
||||
exception.code());
|
||||
}
|
||||
assertTrue(context.containsBeanDefinition("otlpGrpcMetricsService"));
|
||||
assertTrue(context.containsBeanDefinition("alarmGroupReduce"));
|
||||
assertTrue(context.containsBeanDefinition("alarmInhibitReduce"));
|
||||
|
||||
Reference in New Issue
Block a user