fix(shadow): classify source-change refresh as stale
This commit is contained in:
@@ -283,6 +283,17 @@ public class ShadowObservationGateway {
|
||||
metrics.increment(ShadowSkipReason.CONCURRENT);
|
||||
} catch (Exception shadowFailure) {
|
||||
coordinator.failRefresh(ticket, failureDescription(failureStage, shadowFailure));
|
||||
if (isSourceChanged(shadowFailure)) {
|
||||
// SQL 51025 is the stored procedure's optimistic-concurrency guard:
|
||||
// source rows changed while the snapshot was being built. Treat it as
|
||||
// a stale refresh and defer the next attempt; retrying immediately only
|
||||
// creates another collision and incorrectly reports an application error.
|
||||
metrics.increment(ShadowSkipReason.STALE);
|
||||
recoveryNotBefore.set(clock.instant().plus(RECOVERY_COOLDOWN));
|
||||
LOG.info("shadow_refresh_stale warehouse={} snapshotGeneration={} sqlCode=51025",
|
||||
warehouseId, generation);
|
||||
return;
|
||||
}
|
||||
metrics.recordError(failureStage);
|
||||
warnFailure("CONTROL", generation, failureStage, shadowFailure, startedNanos);
|
||||
}
|
||||
@@ -290,6 +301,21 @@ public class ShadowObservationGateway {
|
||||
recoveryNotBefore.set(clock.instant().plus(RECOVERY_COOLDOWN));
|
||||
}
|
||||
|
||||
private boolean isSourceChanged(Throwable failure) {
|
||||
Throwable cause = failure;
|
||||
for (int depth = 0; depth < 8 && cause != null; depth++) {
|
||||
if (cause instanceof SQLException sql && sql.getErrorCode() == 51025) {
|
||||
return true;
|
||||
}
|
||||
Throwable next = cause.getCause();
|
||||
if (next == cause) {
|
||||
break;
|
||||
}
|
||||
cause = next;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private Optional<ShadowSnapshotCoordinator.RefreshTicket> tryBeginRefreshAfterQuietWindow()
|
||||
throws InterruptedException {
|
||||
if (coordinator.activeLegacyCount() != 0) {
|
||||
|
||||
@@ -9,6 +9,7 @@ import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.time.ZoneId;
|
||||
import java.time.ZoneOffset;
|
||||
import java.sql.SQLException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ArrayBlockingQueue;
|
||||
@@ -392,6 +393,24 @@ class ShadowObservationGatewayTest {
|
||||
assertThat(recovered.await(1, TimeUnit.SECONDS)).isTrue();
|
||||
}
|
||||
|
||||
@Test
|
||||
void sourceChangedRefreshIsClassifiedAsStaleAndDoesNotRetry() throws Exception {
|
||||
ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class);
|
||||
ShadowInputMapper mapper = mock(ShadowInputMapper.class);
|
||||
SQLException sourceChanged = new SQLException("source changed", "S0001", 51025);
|
||||
when(engine.refreshSnapshot(anyLong(), eq("INSTANCE-1"))).thenThrow(new RuntimeException(sourceChanged));
|
||||
CountDownLatch staleMetric = metricLatch(mapper, "SKIP_STALE");
|
||||
ShadowObservationGateway gateway = gateway(engine, mapper, coordinator(),
|
||||
new ShadowRuntimeMetrics(), Clock.fixed(NOW, ZoneOffset.UTC), delay -> { });
|
||||
|
||||
gateway.start();
|
||||
|
||||
assertThat(staleMetric.await(1, TimeUnit.SECONDS)).isTrue();
|
||||
verify(engine, timeout(1000).times(1)).refreshSnapshot(anyLong(), eq("INSTANCE-1"));
|
||||
verify(mapper, never()).upsertRuntimeMetric(anyString(), eq("ERROR_REFRESH"), anyLong(),
|
||||
anyLong(), anyLong(), anyLong(), any(Instant.class));
|
||||
}
|
||||
|
||||
private void assertOutboundUsesLease(boolean exactStock) throws Exception {
|
||||
ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class);
|
||||
ShadowInputMapper mapper = mock(ShadowInputMapper.class);
|
||||
|
||||
Reference in New Issue
Block a user