test(ft546): verify read-only shadow operation
This commit is contained in:
@@ -0,0 +1,211 @@
|
||||
package com.ruoyi.shadow.kernel;
|
||||
|
||||
import com.ruoyi.shadow.facts.ShadowFactBatch;
|
||||
import com.ruoyi.shadow.facts.ShadowFactReadException;
|
||||
import com.ruoyi.shadow.facts.ShadowQueueFact;
|
||||
import com.ruoyi.shadow.facts.ShadowReadCursor;
|
||||
import com.ruoyi.shadow.facts.SqlServerShadowFactReader;
|
||||
import com.ruoyi.shadow.startup.ShadowStartupReport;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
|
||||
import java.time.Clock;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.Objects;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.ThreadFactory;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.TimeoutException;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
public final class Ft546ShadowScheduler implements AutoCloseable {
|
||||
private static final Logger LOGGER =
|
||||
LoggerFactory.getLogger(Ft546ShadowScheduler.class);
|
||||
private static final Duration CYCLE_DEADLINE = Duration.ofSeconds(4);
|
||||
private static final Duration CRASH_BACKOFF = Duration.ofSeconds(1);
|
||||
private static final String FINISH_QUEUE_VIEW =
|
||||
"shadow_ro.v_wcs_srgv_queue_finish";
|
||||
|
||||
private final SqlServerShadowFactReader factReader;
|
||||
private final Ft546ShadowCycle shadowCycle;
|
||||
private final ShadowStartupReport statusReport;
|
||||
private final Clock clock;
|
||||
private final Duration cycleDeadline;
|
||||
private final ExecutorService cycleExecutor;
|
||||
private final AtomicBoolean inFlight = new AtomicBoolean();
|
||||
private final AtomicBoolean permanentlyStopped = new AtomicBoolean();
|
||||
private volatile ShadowReadCursor cursor = ShadowReadCursor.initial();
|
||||
private volatile Instant nextAttemptAt = Instant.EPOCH;
|
||||
|
||||
public Ft546ShadowScheduler(SqlServerShadowFactReader factReader,
|
||||
Ft546ShadowCycle shadowCycle,
|
||||
ShadowStartupReport statusReport,
|
||||
Clock clock) {
|
||||
this(factReader, shadowCycle, statusReport, clock, CYCLE_DEADLINE);
|
||||
}
|
||||
|
||||
public Ft546ShadowScheduler(SqlServerShadowFactReader factReader,
|
||||
Ft546ShadowCycle shadowCycle,
|
||||
ShadowStartupReport statusReport,
|
||||
Clock clock,
|
||||
Duration cycleDeadline) {
|
||||
this.factReader = Objects.requireNonNull(factReader, "factReader");
|
||||
this.shadowCycle = Objects.requireNonNull(shadowCycle, "shadowCycle");
|
||||
this.statusReport = Objects.requireNonNull(statusReport,
|
||||
"statusReport");
|
||||
this.clock = Objects.requireNonNull(clock, "clock");
|
||||
this.cycleDeadline = requirePositive(cycleDeadline);
|
||||
this.cycleExecutor = Executors.newSingleThreadExecutor(
|
||||
shadowCycleThreadFactory());
|
||||
}
|
||||
|
||||
@Scheduled(
|
||||
initialDelayString = "${shadow.poll.initial-delay-ms:5000}",
|
||||
fixedDelayString = "${shadow.poll.fixed-delay-ms:5000}")
|
||||
public void runScheduledCycle() {
|
||||
runOnce();
|
||||
}
|
||||
|
||||
public boolean runOnce() {
|
||||
Instant now = clock.instant();
|
||||
if (permanentlyStopped.get() || now.isBefore(nextAttemptAt)
|
||||
|| !inFlight.compareAndSet(false, true)) {
|
||||
return false;
|
||||
}
|
||||
Future<CycleAttempt> future = cycleExecutor.submit(
|
||||
() -> executeCycle(now));
|
||||
try {
|
||||
CycleAttempt attempt = future.get(cycleDeadline.toMillis(),
|
||||
TimeUnit.MILLISECONDS);
|
||||
if ("DEGRADED".equals(attempt.result.getStatusCode())) {
|
||||
permanentlyStopped.set(true);
|
||||
statusReport.markPermanentlyDegraded(
|
||||
attempt.result.getReasonCode(), clock.instant());
|
||||
LOGGER.error("FT546 shadow collection stopped: {}",
|
||||
attempt.result.getReasonCode());
|
||||
} else {
|
||||
cursor = advanceFinishCursor(cursor, attempt.batch);
|
||||
}
|
||||
nextAttemptAt = Instant.EPOCH;
|
||||
return true;
|
||||
} catch (TimeoutException timeout) {
|
||||
future.cancel(true);
|
||||
return stopPermanently("SHADOW_CYCLE_TIMEOUT", timeout);
|
||||
} catch (InterruptedException interrupted) {
|
||||
future.cancel(true);
|
||||
Thread.currentThread().interrupt();
|
||||
return stopPermanently("SHADOW_CYCLE_INTERRUPTED", interrupted);
|
||||
} catch (ExecutionException wrapped) {
|
||||
return handleFailure(wrapped.getCause());
|
||||
} finally {
|
||||
inFlight.set(false);
|
||||
}
|
||||
}
|
||||
|
||||
public boolean isPermanentlyStopped() {
|
||||
return permanentlyStopped.get();
|
||||
}
|
||||
|
||||
public Duration getRemainingBackoff() {
|
||||
Duration remaining = Duration.between(clock.instant(), nextAttemptAt);
|
||||
return remaining.isNegative() ? Duration.ZERO : remaining;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
permanentlyStopped.set(true);
|
||||
cycleExecutor.shutdownNow();
|
||||
}
|
||||
|
||||
private CycleAttempt executeCycle(Instant startedAt) {
|
||||
ShadowFactBatch batch = factReader.read(cursor,
|
||||
startedAt.plus(cycleDeadline));
|
||||
return new CycleAttempt(batch, shadowCycle.run(batch, clock.instant()));
|
||||
}
|
||||
|
||||
private boolean handleFailure(Throwable failure) {
|
||||
if (failure instanceof ShadowFactReadException) {
|
||||
ShadowFactReadException readFailure =
|
||||
(ShadowFactReadException) failure;
|
||||
Instant failedAt = clock.instant();
|
||||
statusReport.markDegraded(readFailure.getReasonCode(), failedAt);
|
||||
nextAttemptAt = failedAt.plus(readFailure.getRetryAfter());
|
||||
LOGGER.warn("FT546 shadow source read failed: {}; retry after {} ms",
|
||||
readFailure.getReasonCode(),
|
||||
readFailure.getRetryAfter().toMillis());
|
||||
return false;
|
||||
}
|
||||
if (failure instanceof Error) {
|
||||
throw (Error) failure;
|
||||
}
|
||||
RuntimeException runtimeFailure = failure instanceof RuntimeException
|
||||
? (RuntimeException) failure
|
||||
: new IllegalStateException("shadow cycle failed", failure);
|
||||
Instant failedAt = clock.instant();
|
||||
statusReport.markDegraded("SHADOW_CYCLE_FAILED", failedAt);
|
||||
nextAttemptAt = failedAt.plus(CRASH_BACKOFF);
|
||||
LOGGER.error("FT546 shadow cycle failed", runtimeFailure);
|
||||
return false;
|
||||
}
|
||||
|
||||
private boolean stopPermanently(String reasonCode, Exception failure) {
|
||||
permanentlyStopped.set(true);
|
||||
nextAttemptAt = Instant.EPOCH;
|
||||
statusReport.markPermanentlyDegraded(reasonCode, clock.instant());
|
||||
LOGGER.error("FT546 shadow collection stopped: {}", reasonCode,
|
||||
failure);
|
||||
return false;
|
||||
}
|
||||
|
||||
private static Duration requirePositive(Duration value) {
|
||||
Duration duration = Objects.requireNonNull(value, "cycleDeadline");
|
||||
if (duration.isZero() || duration.isNegative()
|
||||
|| duration.toMillis() == 0L) {
|
||||
throw new IllegalArgumentException(
|
||||
"cycleDeadline must be at least one millisecond");
|
||||
}
|
||||
return duration;
|
||||
}
|
||||
|
||||
private static ThreadFactory shadowCycleThreadFactory() {
|
||||
return runnable -> {
|
||||
Thread thread = new Thread(runnable, "ft546-shadow-cycle");
|
||||
thread.setDaemon(true);
|
||||
return thread;
|
||||
};
|
||||
}
|
||||
|
||||
private static ShadowReadCursor advanceFinishCursor(
|
||||
ShadowReadCursor current, ShadowFactBatch batch) {
|
||||
Instant latestDate = current.getLastCreateDate();
|
||||
String latestGuid = current.getLastGuid();
|
||||
for (ShadowQueueFact fact : batch.getSourceQueues()) {
|
||||
if (!FINISH_QUEUE_VIEW.equals(fact.getTableName())
|
||||
|| fact.getCreatedAt() == null) {
|
||||
continue;
|
||||
}
|
||||
int dateOrder = fact.getCreatedAt().compareTo(latestDate);
|
||||
if (dateOrder > 0 || (dateOrder == 0
|
||||
&& fact.getGuid().compareTo(latestGuid) > 0)) {
|
||||
latestDate = fact.getCreatedAt();
|
||||
latestGuid = fact.getGuid();
|
||||
}
|
||||
}
|
||||
return new ShadowReadCursor(latestDate, latestGuid);
|
||||
}
|
||||
|
||||
private static final class CycleAttempt {
|
||||
private final ShadowFactBatch batch;
|
||||
private final ShadowCycleResult result;
|
||||
|
||||
private CycleAttempt(ShadowFactBatch batch, ShadowCycleResult result) {
|
||||
this.batch = batch;
|
||||
this.result = result;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user