数据清理

src/main/java/com/adx/tencent/retention/LegacyRetentionJob.java

@@ -0,0 +1,102 @@
+package com.adx.tencent.retention;
+
+import com.adx.tencent.leader.LeaderElection;
+import jakarta.annotation.PreDestroy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.lang.Nullable;
+import org.springframework.stereotype.Component;
+
+import javax.sql.DataSource;
+import java.time.Duration;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+@Component
+@ConditionalOnProperty(prefix = "adx", name = "legacy-retention-enabled", havingValue = "true")
+public class LegacyRetentionJob implements ApplicationRunner {
+    private static final Logger log = LoggerFactory.getLogger(LegacyRetentionJob.class);
+    static final long BATCH_DELAY_MS = 10_000;
+    static final String LOCK_KEY = "adx:lock:retention:adx_proxy:legacy-tables";
+    private final LegacyRetentionStore store;
+    private final LeaderElection leader;
+    private final Waiter waiter;
+    private final AtomicBoolean stopped = new AtomicBoolean();
+    private final ExecutorService executor = Executors.newSingleThreadExecutor(r -> {
+        Thread t = new Thread(r, "legacy-retention");
+        t.setDaemon(true);
+        return t;
+    });
+
+    @Autowired
+    public LegacyRetentionJob(@Nullable DataSource dataSource, LeaderElection leader) {
+        this(dataSource == null ? null : new LegacyRetentionStore(dataSource), leader, LegacyRetentionJob::waitBetweenBatches);
+    }
+
+    LegacyRetentionJob(LegacyRetentionStore store, LeaderElection leader, Waiter waiter) {
+        this.store = store;
+        this.leader = leader;
+        this.waiter = waiter;
+    }
+
+    @Override
+    public void run(ApplicationArguments args) {
+        if (store == null) {
+            log.warn("[LegacyRetention] no datasource; cleanup not started");
+            return;
+        }
+        // Always elect one worker, even when skip-leader-election is set for unrelated jobs.
+        executor.submit(() -> leader.run(LOCK_KEY, Duration.ofMinutes(2), Duration.ofSeconds(30),
+                Duration.ofSeconds(10), stopped::get, this::runLoop,
+                e -> log.error("[LegacyRetention] leader error", e)));
+    }
+
+    void runLoop(LeaderElection.StopSignal signal) {
+        LeaderElection.StopSignal stop = () -> stopped.get() || signal.isStopped() || Thread.currentThread().isInterrupted();
+        LegacyRetentionStore.Table[] tables = LegacyRetentionStore.Table.values();
+        int next = 0;
+        while (!stop.isStopped()) {
+            LegacyRetentionStore.Table table = tables[next];
+            next = (next + 1) % tables.length;
+            try {
+                int deleted = store.deleteBatch(table);
+                if (deleted > 0) log.info("[LegacyRetention] table={} deleted={} retention=30days", table.qualifiedName(), deleted);
+                else log.debug("[LegacyRetention] table={} no expired rows", table.qualifiedName());
+            } catch (Exception e) {
+                // A broken/missing table must not prevent the other table from being cleaned.
+                log.error("[LegacyRetention] table={} batch failed; retry on next turn", table.qualifiedName(), e);
+            }
+            try {
+                waiter.await(BATCH_DELAY_MS, stop);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                return;
+            }
+        }
+    }
+
+    private static void waitBetweenBatches(long millis, LeaderElection.StopSignal stop) throws InterruptedException {
+        long deadline = System.nanoTime() + millis * 1_000_000;
+        while (!stop.isStopped()) {
+            long remaining = deadline - System.nanoTime();
+            if (remaining <= 0) return;
+            Thread.sleep(Math.min(1000, Math.max(1, remaining / 1_000_000)));
+        }
+    }
+
+    @PreDestroy
+    public void close() {
+        stopped.set(true);
+        executor.shutdownNow();
+    }
+
+    @FunctionalInterface
+    interface Waiter {
+        void await(long millis, LeaderElection.StopSignal stop) throws InterruptedException;
+    }
+}

src/main/java/com/adx/tencent/retention/LegacyRetentionStore.java

@@ -0,0 +1,63 @@
+package com.adx.tencent.retention;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.EnumSet;
+
+/** Deletes only the two explicitly requested legacy tables, never media-specific tables. */
+public class LegacyRetentionStore {
+    static final int BATCH_SIZE = 5000;
+    enum Table {
+        BID("ad_bid_events"), TRACKING("tracking_reports");
+        final String name;
+        Table(String name) { this.name = name; }
+        String qualifiedName() { return "`adx_proxy`.`" + name + "`"; }
+    }
+
+    private final DataSource dataSource;
+    private final EnumSet<Table> validated = EnumSet.noneOf(Table.class);
+
+    public LegacyRetentionStore(DataSource dataSource) { this.dataSource = dataSource; }
+
+    public synchronized int deleteBatch(Table table) throws SQLException {
+        try (Connection connection = dataSource.getConnection()) {
+            // One small auto-committed statement per batch, with no transaction spanning the wait.
+            connection.setAutoCommit(true);
+            if (!validated.contains(table)) {
+                validateCreatedAtIndex(connection, table);
+                validated.add(table);
+            }
+            String sql = "DELETE FROM " + table.qualifiedName()
+                    + " WHERE `created_at` < DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL 30 DAY)"
+                    + " ORDER BY `created_at` LIMIT ?";
+            try (PreparedStatement statement = connection.prepareStatement(sql)) {
+                statement.setQueryTimeout(30);
+                statement.setInt(1, BATCH_SIZE);
+                return statement.executeUpdate();
+            }
+        }
+    }
+
+    private void validateCreatedAtIndex(Connection connection, Table table) throws SQLException {
+        // Avoid repeatedly scanning a very large table when the required leading index is absent.
+        String sql = "SELECT COUNT(*) FROM information_schema.STATISTICS s"
+                + " JOIN information_schema.COLUMNS c ON c.TABLE_SCHEMA=s.TABLE_SCHEMA"
+                + " AND c.TABLE_NAME=s.TABLE_NAME AND c.COLUMN_NAME=s.COLUMN_NAME"
+                + " WHERE s.TABLE_SCHEMA='adx_proxy' AND s.TABLE_NAME=?"
+                + " AND s.COLUMN_NAME='created_at' AND s.SEQ_IN_INDEX=1"
+                + " AND c.DATA_TYPE IN ('timestamp','datetime','date')";
+        try (PreparedStatement statement = connection.prepareStatement(sql)) {
+            statement.setQueryTimeout(30);
+            statement.setString(1, table.name);
+            try (ResultSet result = statement.executeQuery()) {
+                if (!result.next() || result.getLong(1) == 0) {
+                    throw new SQLException(table.qualifiedName()
+                            + " requires a date/time created_at column and an index starting with created_at; cleanup skipped");
+                }
+            }
+        }
+    }
+}

src/main/resources/application-dev.yml

@@ -19,6 +19,8 @@ baidu:
     read-timeout-seconds: 30
 
 adx:
+  # Only adx_proxy.ad_bid_events / tracking_reports; 5000 rows per batch, 10s between batches.
+  legacy-retention-enabled: false
   # --- Baidu ADX ---
   baidu-adx-endpoint: "http://adsp.baidu.com/jbssp"
   baidu-conversion-base-url: "https://media-interact.baidu.com"

src/main/resources/application-prod.yml

@@ -19,6 +19,8 @@ baidu:
     read-timeout-seconds: 30
 
 adx:
+  # Only adx_proxy.ad_bid_events / tracking_reports; 5000 rows per batch, 10s between batches.
+  legacy-retention-enabled: ${ADX_LEGACY_RETENTION_ENABLED:true}
   # --- Baidu ADX ---
   baidu-adx-endpoint: "http://adsp.baidu.com/jbssp"
   baidu-conversion-base-url: "https://media-interact.baidu.com"

src/main/resources/application-test.yml

@@ -19,6 +19,8 @@ baidu:
     read-timeout-seconds: 30
 
 adx:
+  # Only adx_proxy.ad_bid_events / tracking_reports; 5000 rows per batch, 10s between batches.
+  legacy-retention-enabled: false
   # --- Baidu ADX ---
   baidu-adx-endpoint: "http://adsp.baidu.com/jbssp"
   baidu-conversion-base-url: "https://media-interact-test.baidu.com"

src/test/java/com/adx/tencent/retention/LegacyRetentionJobTest.java

@@ -0,0 +1,61 @@
+package com.adx.tencent.retention;
+
+import com.adx.tencent.leader.LeaderElection;
+import org.junit.jupiter.api.Test;
+import java.sql.SQLException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+class LegacyRetentionJobTest {
+    @Test void alternatesTablesAndWaitsTenSecondsAfterEveryBatchIncludingFailures() throws Exception {
+        LegacyRetentionStore store = mock(LegacyRetentionStore.class);
+        List<String> events = new ArrayList<>();
+        when(store.deleteBatch(any())).thenAnswer(i -> {
+            LegacyRetentionStore.Table table = i.getArgument(0);
+            events.add(table.name());
+            if (table == LegacyRetentionStore.Table.BID) throw new SQLException("failed table");
+            return 5000;
+        });
+        AtomicBoolean stop = new AtomicBoolean();
+        LegacyRetentionJob job = new LegacyRetentionJob(store, mock(LeaderElection.class), (millis, signal) -> {
+            events.add("wait:" + millis);
+            if (events.size() == 8) stop.set(true);
+        });
+        try {
+            job.runLoop(stop::get);
+            assertEquals(List.of("BID", "wait:10000", "TRACKING", "wait:10000",
+                    "BID", "wait:10000", "TRACKING", "wait:10000"), events);
+        } finally { job.close(); }
+    }
+
+    @Test void losingLeadershipPreventsAnyNewDeletion() {
+        LegacyRetentionStore store = mock(LegacyRetentionStore.class);
+        LegacyRetentionJob job = new LegacyRetentionJob(store, mock(LeaderElection.class), (ms, stop) -> fail());
+        try { job.runLoop(() -> true); verifyNoInteractions(store); }
+        finally { job.close(); }
+    }
+
+    @Test void startsOnlyThroughIndependentLeaderLock() {
+        LegacyRetentionStore store = mock(LegacyRetentionStore.class);
+        LeaderElection leader = mock(LeaderElection.class);
+        LegacyRetentionJob job = new LegacyRetentionJob(store, leader, (ms, stop) -> fail());
+        try {
+            job.run(null);
+            verify(leader, timeout(2000)).run(eq(LegacyRetentionJob.LOCK_KEY), eq(Duration.ofMinutes(2)),
+                    eq(Duration.ofSeconds(30)), eq(Duration.ofSeconds(10)), any(), any(), any());
+            verifyNoInteractions(store); // No deletion before the leader invokes the worker.
+        } finally { job.close(); }
+    }
+
+    @Test void noDataSourceDoesNotStartWorker() {
+        LeaderElection leader = mock(LeaderElection.class);
+        LegacyRetentionJob job = new LegacyRetentionJob(null, leader, (ms, stop) -> fail());
+        try { job.run(null); verifyNoInteractions(leader); }
+        finally { job.close(); }
+    }
+}

src/test/java/com/adx/tencent/retention/LegacyRetentionStoreTest.java

@@ -0,0 +1,60 @@
+package com.adx.tencent.retention;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import javax.sql.DataSource;
+import java.sql.*;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+class LegacyRetentionStoreTest {
+    final DataSource ds = mock(DataSource.class);
+    final Connection connection = mock(Connection.class);
+    final PreparedStatement metadata = mock(PreparedStatement.class);
+    final PreparedStatement delete = mock(PreparedStatement.class);
+    final ResultSet result = mock(ResultSet.class);
+
+    LegacyRetentionStore store(boolean indexExists) throws Exception {
+        when(ds.getConnection()).thenReturn(connection);
+        when(connection.prepareStatement(startsWith("SELECT COUNT(*)"))).thenReturn(metadata);
+        when(connection.prepareStatement(startsWith("DELETE FROM"))).thenReturn(delete);
+        when(metadata.executeQuery()).thenReturn(result);
+        when(result.next()).thenReturn(true);
+        when(result.getLong(1)).thenReturn(indexExists ? 1L : 0L);
+        when(delete.executeUpdate()).thenReturn(5000);
+        return new LegacyRetentionStore(ds);
+    }
+
+    @ParameterizedTest @EnumSource(LegacyRetentionStore.Table.class)
+    void onlyDeletesExpiredRowsInExactTableWithBoundedAutoCommittedBatch(LegacyRetentionStore.Table table) throws Exception {
+        assertEquals(5000, store(true).deleteBatch(table));
+        verify(connection).prepareStatement("DELETE FROM `adx_proxy`.`" + table.name + "`"
+                + " WHERE `created_at` < DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL 30 DAY)"
+                + " ORDER BY `created_at` LIMIT ?");
+        verify(metadata).setString(1, table.name);
+        verify(delete).setInt(1, 5000);
+        verify(delete).setQueryTimeout(30);
+        verify(connection).setAutoCommit(true);
+        verify(delete).close(); verify(result).close(); verify(metadata).close(); verify(connection).close();
+    }
+
+    @Test void missingIndexBlocksDeletionAndIsRecheckedNextTime() throws Exception {
+        LegacyRetentionStore store = store(false);
+        assertThrows(SQLException.class, () -> store.deleteBatch(LegacyRetentionStore.Table.BID));
+        verify(delete, never()).executeUpdate();
+        when(result.getLong(1)).thenReturn(1L);
+        assertEquals(5000, store.deleteBatch(LegacyRetentionStore.Table.BID));
+        verify(metadata, times(2)).executeQuery();
+    }
+
+    @Test void failedDeleteClosesConnectionAndNextBatchCanRetry() throws Exception {
+        LegacyRetentionStore store = store(true);
+        when(delete.executeUpdate()).thenThrow(new SQLException("timeout")).thenReturn(0);
+        assertThrows(SQLException.class, () -> store.deleteBatch(LegacyRetentionStore.Table.TRACKING));
+        assertEquals(0, store.deleteBatch(LegacyRetentionStore.Table.TRACKING));
+        verify(connection, times(2)).close();
+        verify(delete, times(2)).close();
+    }
+}