Distinct from {@code org.apache.gravitino.hook.PolicyHookDispatcher}, which decorates the + * policy API after a successful manager call (for example ownership). This SPI runs in {@code + * PolicyMetaService} on the open SQL session so extension writes commit or roll back with the + * policy row. + * + *
Implementations must use that connection, and must not commit, roll back, or close it. An + * exception rolls the policy write back together with the hook's own writes. + * + *
Defaults are no-ops so an extension only overrides the points it needs. + * + *
{@code insertPolicy(..., overwritten = true)} that replaces an existing row fires the update + * lifecycle ({@link #beforeUpdate} / {@link #afterUpdate}), not persist. A create, or an overwrite + * that inserts because no row was found, fires persist. + */ +public interface PolicyLifecycleHook { + + /** + * About to insert the policy row in the current transaction. + * + * @param policy the policy about to be stored + */ + default void beforePersist(PolicyEntity policy) {} + + /** + * The policy row has been inserted in the current transaction. + * + * @param policy the policy as stored, including id, enabled flag, and content + */ + default void afterPersist(PolicyEntity policy) {} + + /** + * About to update the policy row in the current transaction. + * + *
Also used when {@code insertPolicy} overwrites an existing row by name or id. + * + * @param policy the policy about to be stored (id is the retained row id on overwrite) + */ + default void beforeUpdate(PolicyEntity policy) {} + + /** + * The policy row has been updated in the current transaction. + * + *
Also used when {@code insertPolicy} overwrites an existing row by name or id. + * + * @param policy the policy as stored, including the retained id, enabled flag, and content + */ + default void afterUpdate(PolicyEntity policy) {} + + /** + * The policy row is about to be deleted in the current transaction. + * + * @param policyId persisted policy id + */ + default void beforeRemove(long policyId) {} + + /** + * The policy row has been deleted in the current transaction. + * + * @param policyId persisted policy id + */ + default void afterRemove(long policyId) {} +} diff --git a/core/src/main/java/org/apache/gravitino/policy/hook/PolicyLifecycleHooks.java b/core/src/main/java/org/apache/gravitino/policy/hook/PolicyLifecycleHooks.java new file mode 100644 index 00000000000..9e6f7a9c3c8 --- /dev/null +++ b/core/src/main/java/org/apache/gravitino/policy/hook/PolicyLifecycleHooks.java @@ -0,0 +1,66 @@ +/* + * 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 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.policy.hook; + +import java.util.concurrent.atomic.AtomicReference; +import javax.annotation.Nullable; + +/** + * Holds the optional {@link PolicyLifecycleHook} registered by an extension. + * + *
This is a storage-layer SPI invoked from {@code PolicyMetaService} inside the open policy
+ * transaction. It is not the same as {@code org.apache.gravitino.hook.PolicyHookDispatcher}, which
+ * wraps the policy API for owner side effects after a successful manager call.
+ */
+public final class PolicyLifecycleHooks {
+
+ private static final AtomicReference A plain create (or an overwrite that finds no row) uses persist hooks. An overwrite that
+ * replaces an existing row uses update hooks and passes the entity as stored — including the
+ * retained policy id when the caller supplied a different id for the same name.
*
* An overwrite is not an upsert any more: the existing row is located and locked first, and
* the replacement is written as the next version of that row, so the snapshot history survives
@@ -378,7 +389,9 @@ private void lockMetalakeForPolicyCreate(MetalakePO observedMetalakePO) {
private void insertPolicyWithoutCommit(
PolicyEntity policyEntity, PolicyPO initializedPolicyPO, boolean overwritten) {
if (!overwritten) {
+ beforePersist(policyEntity);
insertNewPolicyWithoutCommit(initializedPolicyPO);
+ afterPersist(policyEntity);
return;
}
@@ -392,16 +405,22 @@ private void insertPolicyWithoutCommit(
"The policy ID %s is reserved by a deleted policy; use a new ID",
initializedPolicyPO.getPolicyId());
}
+ beforePersist(policyEntity);
insertNewPolicyWithoutCommit(initializedPolicyPO);
+ afterPersist(policyEntity);
return;
}
PolicyPO replacementPolicyPO =
POConverters.updatePolicyPOWithVersion(existingPolicyPO, initializedPolicyPO);
+ PolicyEntity storedPolicy =
+ POConverters.fromPolicyPO(replacementPolicyPO, policyEntity.namespace());
NameIdentifier observedIdentifier =
NameIdentifier.of(policyEntity.namespace(), existingPolicyPO.getPolicyName());
+ beforeUpdate(storedPolicy);
updatePolicyRootWithVersion(observedIdentifier, existingPolicyPO, replacementPolicyPO);
insertPolicyVersionIfAllocated(existingPolicyPO, replacementPolicyPO);
+ afterUpdate(storedPolicy);
}
/**
@@ -570,4 +589,46 @@ private PolicyPO getPolicyPOByMetalakeAndName(String metalakeName, String policy
}
return policyPO;
}
+
+ private static void beforePersist(PolicyEntity policy) {
+ PolicyLifecycleHook hook = PolicyLifecycleHooks.get();
+ if (hook != null) {
+ hook.beforePersist(policy);
+ }
+ }
+
+ private static void afterPersist(PolicyEntity policy) {
+ PolicyLifecycleHook hook = PolicyLifecycleHooks.get();
+ if (hook != null) {
+ hook.afterPersist(policy);
+ }
+ }
+
+ private static void beforeUpdate(PolicyEntity policy) {
+ PolicyLifecycleHook hook = PolicyLifecycleHooks.get();
+ if (hook != null) {
+ hook.beforeUpdate(policy);
+ }
+ }
+
+ private static void afterUpdate(PolicyEntity policy) {
+ PolicyLifecycleHook hook = PolicyLifecycleHooks.get();
+ if (hook != null) {
+ hook.afterUpdate(policy);
+ }
+ }
+
+ private static void beforeRemove(long policyId) {
+ PolicyLifecycleHook hook = PolicyLifecycleHooks.get();
+ if (hook != null) {
+ hook.beforeRemove(policyId);
+ }
+ }
+
+ private static void afterRemove(long policyId) {
+ PolicyLifecycleHook hook = PolicyLifecycleHooks.get();
+ if (hook != null) {
+ hook.afterRemove(policyId);
+ }
+ }
}
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/session/SqlSessions.java b/core/src/main/java/org/apache/gravitino/storage/relational/session/SqlSessions.java
index 48a10324b5d..2ebe3d30b4c 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/session/SqlSessions.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/session/SqlSessions.java
@@ -20,6 +20,8 @@
package org.apache.gravitino.storage.relational.session;
import com.google.common.annotations.VisibleForTesting;
+import java.sql.Connection;
+import java.sql.SQLException;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.TransactionIsolationLevel;
@@ -112,6 +114,23 @@ public static boolean hasActiveSession() {
return sessions.get() != null;
}
+ /**
+ * Returns the connection of the current thread's open session without changing the session count.
+ *
+ * Callers must not commit, roll back, or close the connection. The session owner commits it.
+ *
+ * @return the connection bound to the current session
+ * @throws IllegalStateException if the current thread has no open session
+ * @throws SQLException if the session cannot provide its connection
+ */
+ public static Connection currentConnection() throws SQLException {
+ SqlSession sqlSession = sessions.get();
+ if (sqlSession == null) {
+ throw new IllegalStateException("No active SQL session on the current thread");
+ }
+ return sqlSession.getConnection();
+ }
+
private static void handleSessionClose(boolean commit, boolean rollback) {
SqlSession sqlSession = sessions.get();
if (sqlSession == null) {
@@ -124,7 +143,11 @@ private static void handleSessionClose(boolean commit, boolean rollback) {
if (commit) {
sqlSession.commit();
} else if (rollback) {
- sqlSession.rollback();
+ // Force the JDBC rollback. A policy lifecycle hook can write on this connection without
+ // going
+ // through a mapped statement, and MyBatis skips rollback when the session looks clean.
+ // Closing that session then turns auto-commit back on, which would commit the hook write.
+ sqlSession.rollback(true);
}
} finally {
try {
diff --git a/core/src/test/java/org/apache/gravitino/policy/hook/TestPolicyLifecycleHooks.java b/core/src/test/java/org/apache/gravitino/policy/hook/TestPolicyLifecycleHooks.java
new file mode 100644
index 00000000000..6d802b91677
--- /dev/null
+++ b/core/src/test/java/org/apache/gravitino/policy/hook/TestPolicyLifecycleHooks.java
@@ -0,0 +1,59 @@
+/*
+ * 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 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.policy.hook;
+
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+public class TestPolicyLifecycleHooks {
+
+ private final PolicyLifecycleHook hook = new PolicyLifecycleHook() {};
+
+ @AfterEach
+ void tearDown() {
+ PolicyLifecycleHooks.unregister(hook);
+ }
+
+ @Test
+ void testRegisterAndUnregister() {
+ PolicyLifecycleHooks.register(hook);
+ assertSame(hook, PolicyLifecycleHooks.get());
+ PolicyLifecycleHooks.unregister(hook);
+ assertNull(PolicyLifecycleHooks.get());
+ }
+
+ @Test
+ void testDoubleRegisterRejected() {
+ PolicyLifecycleHooks.register(hook);
+ PolicyLifecycleHook other = new PolicyLifecycleHook() {};
+ assertThrows(IllegalStateException.class, () -> PolicyLifecycleHooks.register(other));
+ assertSame(hook, PolicyLifecycleHooks.get());
+ }
+
+ @Test
+ void testUnregisterWrongHookIsNoOp() {
+ PolicyLifecycleHooks.register(hook);
+ PolicyLifecycleHooks.unregister(new PolicyLifecycleHook() {});
+ assertSame(hook, PolicyLifecycleHooks.get());
+ }
+}
diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestPolicyMetaService.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestPolicyMetaService.java
index 74dc33e67b0..8a89cc9014b 100644
--- a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestPolicyMetaService.java
+++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestPolicyMetaService.java
@@ -34,6 +34,7 @@
import java.sql.SQLException;
import java.sql.Statement;
import java.time.Instant;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -65,6 +66,8 @@
import org.apache.gravitino.policy.Policy;
import org.apache.gravitino.policy.PolicyContent;
import org.apache.gravitino.policy.PolicyContents;
+import org.apache.gravitino.policy.hook.PolicyLifecycleHook;
+import org.apache.gravitino.policy.hook.PolicyLifecycleHooks;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.relational.TestJDBCBackend;
import org.apache.gravitino.storage.relational.mapper.PolicyMetaMapper;
@@ -93,6 +96,159 @@ public class TestPolicyMetaService extends TestJDBCBackend {
private final PolicyContent content =
PolicyContents.custom(ImmutableMap.of("filed1", 123), supportedObjectTypes, null);
+ @TestTemplate
+ public void testLifecycleHookFailureRollsBackInsert() throws IOException {
+ BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME);
+ PolicyLifecycleHook hook =
+ new PolicyLifecycleHook() {
+ @Override
+ public void afterPersist(PolicyEntity policy) {
+ throw new IllegalArgumentException("bad cron");
+ }
+ };
+ PolicyLifecycleHooks.register(hook);
+ try {
+ PolicyEntity policyEntity =
+ PolicyEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("hook-rollback")
+ .withNamespace(NamespaceUtil.ofPolicy(metalake.name()))
+ .withComment("comment")
+ .withPolicyType(Policy.BuiltInType.CUSTOM)
+ .withContent(content)
+ .withEnabled(true)
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ PolicyMetaService policyMetaService = PolicyMetaService.getInstance();
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> policyMetaService.insertPolicy(policyEntity, false));
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ policyMetaService.getPolicyByIdentifier(
+ NameIdentifierUtil.ofPolicy(metalake.name(), "hook-rollback")));
+ } finally {
+ PolicyLifecycleHooks.unregister(hook);
+ }
+ }
+
+ @TestTemplate
+ public void testLifecycleHookFailureRollsBackUpdate() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ PolicyMetaService policyMetaService = PolicyMetaService.getInstance();
+ PolicyEntity policy =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy(METALAKE_NAME),
+ "hook-update-rollback",
+ AUDIT_INFO);
+ policyMetaService.insertPolicy(policy, false);
+
+ PolicyLifecycleHook hook =
+ new PolicyLifecycleHook() {
+ @Override
+ public void afterUpdate(PolicyEntity updated) {
+ throw new IllegalArgumentException("bad cron on update");
+ }
+ };
+ PolicyLifecycleHooks.register(hook);
+ try {
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ policyMetaService.updatePolicy(
+ policy.nameIdentifier(),
+ entity -> copyPolicy((PolicyEntity) entity, policy.name(), "should-not-stick")));
+ assertEquals(
+ policy.comment(),
+ policyMetaService.getPolicyByIdentifier(policy.nameIdentifier()).comment());
+ } finally {
+ PolicyLifecycleHooks.unregister(hook);
+ }
+ }
+
+ @TestTemplate
+ public void testLifecycleHookFailureRollsBackDelete() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ PolicyMetaService policyMetaService = PolicyMetaService.getInstance();
+ PolicyEntity policy =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy(METALAKE_NAME),
+ "hook-delete-rollback",
+ AUDIT_INFO);
+ policyMetaService.insertPolicy(policy, false);
+
+ PolicyLifecycleHook hook =
+ new PolicyLifecycleHook() {
+ @Override
+ public void beforeRemove(long policyId) {
+ throw new IllegalArgumentException("bad cron on delete");
+ }
+ };
+ PolicyLifecycleHooks.register(hook);
+ try {
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> policyMetaService.deletePolicy(policy.nameIdentifier()));
+ assertTrue(backend.exists(policy.nameIdentifier(), Entity.EntityType.POLICY));
+ } finally {
+ PolicyLifecycleHooks.unregister(hook);
+ }
+ }
+
+ @TestTemplate
+ public void testOverwriteFiresUpdateHooksWithRetainedPolicyId() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ PolicyMetaService policyMetaService = PolicyMetaService.getInstance();
+ PolicyEntity existing =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy(METALAKE_NAME),
+ "hook-overwrite-id",
+ AUDIT_INFO);
+ policyMetaService.insertPolicy(existing, false);
+
+ List An enabled maintenance policy with a non-blank {@code timeTrigger} is scheduled. Disable, a
+ * blank trigger, and a non-maintenance write remove the row. Delete removes the row by policy id.
+ * An invalid cron throws on write so the policy change rolls back; on startup reconcile the expand
+ * row is removed so scheduler state is corrected while the bad policy content remains to be edited.
+ */
+public final class PolicyExpandLifecycleHook implements PolicyLifecycleHook {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PolicyExpandLifecycleHook.class);
+
+ private final PolicyExpandRegistrar registrar;
+
+ /**
+ * @param registrar registers policy-expand task rows on the current SQL session
+ */
+ public PolicyExpandLifecycleHook(PolicyExpandRegistrar registrar) {
+ this.registrar = registrar;
+ }
+
+ @Override
+ public void afterPersist(PolicyEntity policy) {
+ syncExpandRow(policy);
+ }
+
+ @Override
+ public void afterUpdate(PolicyEntity policy) {
+ syncExpandRow(policy);
+ }
+
+ @Override
+ public void beforeRemove(long policyId) {
+ registrar.remove(policyId);
+ }
+
+ /**
+ * Aligns the {@code policy-expand} task for a stored policy (startup reconcile).
+ *
+ * Only mutates expand tasks: inserts when missing; reschedules a future due that no
+ * longer matches the policy cron; leaves an overdue due (< now) alone so the expand pool can
+ * catch up after downtime. Removes the expand task when the policy must not run or the cron is
+ * invalid. Never deletes the policy itself.
+ *
+ * @param policy persisted policy
+ */
+ public void align(PolicyEntity policy) {
+ if (!isMaintenance(policy.content())) {
+ registrar.remove(policy.id());
+ return;
+ }
+ String trigger = timeTrigger(policy.content());
+ if (!policy.enabled() || trigger == null || trigger.isBlank()) {
+ registrar.remove(policy.id());
+ return;
+ }
+ Instant expectedNext;
+ try {
+ expectedNext = PolicyExpandRegistrar.nextExecution(trigger);
+ } catch (IllegalArgumentException e) {
+ registrar.remove(policy.id());
+ LOG.error(
+ "Cleared policy-expand task for policy id {} name {} with invalid timeTrigger; fix the policy content",
+ policy.id(),
+ policy.name(),
+ e);
+ return;
+ }
+ Optional Reloads the policy and either removes the row or moves it to the next cron fire. An empty
+ * completion would leave the row picked, and db-scheduler then rejects alter and delete of that
+ * policy.
+ *
+ * @param instanceId db-scheduler task instance id, the policy id
+ * @param complete the pick that just finished
+ * @param operations completion actions for that pick
+ */
+ public static void completePick(
+ String instanceId, ExecutionComplete complete, ExecutionOperations Only {@code scheduled_expand_tasks} rows are touched — never policy rows or other scheduler
+ * tables. Lifecycle hooks cover later writes; startup must still converge existing state:
+ *
+ * The instance id is the policy id. A blank {@code timeTrigger}, or a disabled policy, removes
+ * the row. Handlers that fan the row out into cron jobs are out of scope here.
+ *
+ * {@code timeTrigger} is a six-field Quartz cron. This class decides whether the expression has
+ * a next fire time; policy content does not parse cron itself.
+ *
+ * {@link #joiningCurrentTransaction()} writes through the open MyBatis session. The client must
+ * not commit or close that connection, so the policy transaction and the expand row commit
+ * together.
+ */
+public final class PolicyExpandRegistrar {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PolicyExpandRegistrar.class);
+
+ /** Attempts while a pick still owns the expand row. */
+ private static final int EXECUTING_ATTEMPTS = 10;
+
+ /** Pause between attempts, so a short pick can finish inside the policy transaction. */
+ private static final long EXECUTING_WAIT_MS = 50L;
+
+ private final DataSource dataSource;
+
+ private PolicyExpandRegistrar(DataSource dataSource) {
+ this.dataSource = dataSource;
+ }
+
+ /**
+ * Opens a client that writes on the current thread's open SQL session.
+ *
+ * The db-scheduler client is created on each call, because building it borrows a connection.
+ * That connection is the open session, so this registrar must be used inside a policy
+ * transaction.
+ *
+ * @return registrar that inserts, moves, and deletes expand rows in the caller's transaction
+ */
+ public static PolicyExpandRegistrar joiningCurrentTransaction() {
+ return new PolicyExpandRegistrar(new SessionDataSource());
+ }
+
+ /**
+ * Opens a client that reads committed rows from the entity-store pool.
+ *
+ * @param dataSource relational JDBC pool shared with the entity store
+ * @return registrar that can read expand rows after the policy transaction commits
+ */
+ public static PolicyExpandRegistrar forReads(DataSource dataSource) {
+ return new PolicyExpandRegistrar(dataSource);
+ }
+
+ private SchedulerClient client() {
+ return SchedulerClient.Builder.create(
+ dataSource, TableMaintenanceSchedulerPools.policyExpandTask())
+ .tableName(TableMaintenanceSchedulerPools.EXPAND_TASKS_TABLE)
+ .build();
+ }
+
+ /**
+ * Inserts or moves the expand row to the next cron fire time, or removes it when the policy
+ * should not run.
+ *
+ * @param policyId persisted policy id, used as the db-scheduler task instance
+ * @param enabled whether the policy is enabled
+ * @param timeTrigger cron expression, or {@code null} when the policy has no time trigger
+ */
+ public void sync(long policyId, boolean enabled, @Nullable String timeTrigger) {
+ TaskInstanceId instanceId = instanceId(policyId);
+ if (!enabled || timeTrigger == null || timeTrigger.isBlank()) {
+ cancelIfPresent(instanceId);
+ return;
+ }
+ scheduleAt(policyId, nextExecution(timeTrigger));
+ }
+
+ /**
+ * Inserts or moves the expand row to an explicit due time.
+ *
+ * @param policyId persisted policy id
+ * @param due execution time to store
+ */
+ @VisibleForTesting
+ void scheduleAt(long policyId, Instant due) {
+ runDespiteExecution(() -> scheduleOnce(policyId, due));
+ }
+
+ /**
+ * Deletes the expand row for a dropped policy.
+ *
+ * @param policyId persisted policy id
+ */
+ public void remove(long policyId) {
+ cancelIfPresent(instanceId(policyId));
+ }
+
+ /**
+ * @param policyId persisted policy id
+ * @return next execution time of the expand row, or empty when the row is absent
+ */
+ Optional A pick rejects both reschedule and cancel. Failing the attempt here would roll the policy
+ * transaction back. After the retry window the in-flight execution applies the committed policy
+ * from {@link PolicyExpandLifecycleHook#completePick}.
+ *
+ * @param action one scheduler write
+ */
+ private static void runDespiteExecution(Runnable action) {
+ for (int attempt = 1; ; attempt++) {
+ try {
+ action.run();
+ return;
+ } catch (TaskInstanceCurrentlyExecutingException e) {
+ if (attempt >= EXECUTING_ATTEMPTS) {
+ LOG.warn(
+ "policy-expand row is still executing; leaving the in-flight pick to apply the policy",
+ e);
+ return;
+ }
+ try {
+ Thread.sleep(EXECUTING_WAIT_MS);
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ e.addSuppressed(interrupted);
+ throw e;
+ }
+ }
+ }
+ }
+
+ /**
+ * Hands db-scheduler the connection of the open SQL session. {@code close}, {@code commit}, and
+ * {@code rollback} are ignored so the session owner commits the expand row with the policy.
+ */
+ private static final class SessionDataSource implements DataSource {
+
+ @Override
+ public Connection getConnection() throws SQLException {
+ return nonClosing(SqlSessions.currentConnection());
+ }
+
+ @Override
+ public Connection getConnection(String username, String password) throws SQLException {
+ return getConnection();
+ }
+
+ @Override
+ public PrintWriter getLogWriter() {
+ return null;
+ }
+
+ @Override
+ public void setLogWriter(PrintWriter out) {}
+
+ @Override
+ public void setLoginTimeout(int seconds) {}
+
+ @Override
+ public int getLoginTimeout() {
+ return 0;
+ }
+
+ @Override
+ public java.util.logging.Logger getParentLogger() throws SQLFeatureNotSupportedException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public db-scheduler requires a dedicated {@code tableName} per Scheduler instance for multi-pool
* isolation; sharing one physical table across pools is unsupported (see design §5.2).
*
- * Task handlers are intentionally minimal in this change; submit and occupancy logic land in
- * follow-up PRs.
+ * Cron and commit pools poll. {@code policy-expand} rows are written with a {@link
+ * com.github.kagkarlsson.scheduler.SchedulerClient}; expand polling starts with fan-out. {@link
+ * #policyExpandTask()} reschedules to the next cron after a pick, so a run does not leave the row
+ * picked.
*/
public class TableMaintenanceSchedulerPools implements AutoCloseable {
@@ -65,7 +68,11 @@ public class TableMaintenanceSchedulerPools implements AutoCloseable {
private final List The expand pool stays stopped. Due {@code policy-expand} rows would otherwise be picked by
+ * an empty handler and block policy alter and delete. Expand polling starts when fan-out can both
+ * run and reschedule the row.
*
* If any pool fails to start, already-started pools are stopped and this instance returns to
* the idle state so {@link #start} may be retried.
@@ -79,12 +86,6 @@ public void start(DataSource dataSource, TableMaintenanceConfig config) {
List {@link #start} does not poll this task. When a run does happen, completion reschedules to
+ * the next cron or removes the row, so the execution does not stay picked.
+ *
+ * @return the {@code policy-expand} task
+ */
+ public static Task TMS does not register public REST resources in this change; the feature only owns server
- * lifecycle for in-process scheduling. Pools stop on Jersey {@code DESTROY_FINISHED}, which runs
- * during Jetty shutdown before {@code GravitinoEnv} closes the entity-store {@code DataSource}.
+ * lifecycle for in-process scheduling. Cron and commit pools poll. The expand pool stays stopped,
+ * and a policy lifecycle hook writes the matching {@code policy-expand} row on the same transaction
+ * as create, alter, enable, disable, and delete. On start, existing maintenance policies are
+ * reconciled into expand rows so upgrade and first-enable paths do not wait for an edit. Pools stop
+ * on Jersey {@code DESTROY_FINISHED}, which runs during Jetty shutdown before {@code GravitinoEnv}
+ * closes the entity-store {@code DataSource}.
*
- * After {@link #stopSchedulers()}, a later {@link #configure} may start pools again and
- * re-register the destroy listener (needed for tests / embedded restarts in the same JVM).
+ * After {@link #stopSchedulers()}, a later {@link #configure} may start pools again, re-register
+ * the destroy listener, and reconcile again (needed for tests / embedded restarts in the same JVM).
*/
@Provider
public class TableMaintenanceRESTFeature implements Feature {
@@ -58,6 +66,9 @@ public class TableMaintenanceRESTFeature implements Feature {
private static final AtomicReference Unregisters the lifecycle hook first so later policy writes skip expand sync. In-flight
+ * transactions that already passed the hook may still finish. A later {@link #configure}
+ * reconciles expand rows again from stored policies.
+ */
public static void stopSchedulers() {
+ unregisterLifecycleHook();
TableMaintenanceSchedulerPools pools = SCHEDULER_POOLS.getAndSet(null);
if (pools != null) {
pools.stop();
@@ -96,6 +124,26 @@ public static void stopSchedulers() {
}
}
+ private static void registerLifecycleHook() {
+ PolicyExpandLifecycleHook hook =
+ new PolicyExpandLifecycleHook(PolicyExpandRegistrar.joiningCurrentTransaction());
+ PolicyLifecycleHooks.register(hook);
+ LIFECYCLE_HOOK.set(hook);
+ }
+
+ private static void reconcileExistingPolicies() {
+ PolicyExpandRegistrar registrar = PolicyExpandRegistrar.joiningCurrentTransaction();
+ PolicyExpandLifecycleHook aligner = new PolicyExpandLifecycleHook(registrar);
+ new PolicyExpandReconciler(aligner, registrar).reconcileAll();
+ }
+
+ private static void unregisterLifecycleHook() {
+ PolicyExpandLifecycleHook hook = LIFECYCLE_HOOK.getAndSet(null);
+ if (hook != null) {
+ PolicyLifecycleHooks.unregister(hook);
+ }
+ }
+
private static DataSource resolveEntityStoreDataSource() {
Object dataSource =
SqlSessionFactoryHelper.getInstance()
diff --git a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/policy/TestPolicyExpandLifecycleHook.java b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/policy/TestPolicyExpandLifecycleHook.java
new file mode 100644
index 00000000000..69336594a12
--- /dev/null
+++ b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/policy/TestPolicyExpandLifecycleHook.java
@@ -0,0 +1,135 @@
+/*
+ * 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 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.maintenance.policy;
+
+import java.util.Map;
+import java.util.Set;
+import javax.sql.DataSource;
+import org.apache.gravitino.MetadataObject;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.PolicyEntity;
+import org.apache.gravitino.policy.IcebergDataCompactionContent;
+import org.apache.gravitino.policy.Policy;
+import org.apache.gravitino.policy.PolicyContent;
+import org.apache.gravitino.policy.PolicyContents;
+import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
+import org.apache.gravitino.utils.NamespaceUtil;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.TestTemplate;
+
+class TestPolicyExpandLifecycleHook extends TestJDBCBackend {
+
+ private static final String CRON = "0 0 2 * * ?";
+
+ @TestTemplate
+ void testEnabledMaintenancePolicyIsScheduled() {
+ PolicyExpandRegistrar writer = PolicyExpandRegistrar.joiningCurrentTransaction();
+ PolicyExpandRegistrar reader = PolicyExpandRegistrar.forReads(entityStoreDataSource());
+ PolicyExpandLifecycleHook hook = new PolicyExpandLifecycleHook(writer);
+
+ SessionUtils.doMultipleWithCommit(() -> hook.afterPersist(policy(42L, true, compaction(CRON))));
+
+ Assertions.assertTrue(reader.scheduledExecutionTime(42L).isPresent());
+ }
+
+ @TestTemplate
+ void testDisabledMaintenancePolicyRemovesTheRow() {
+ PolicyExpandRegistrar writer = PolicyExpandRegistrar.joiningCurrentTransaction();
+ PolicyExpandRegistrar reader = PolicyExpandRegistrar.forReads(entityStoreDataSource());
+ PolicyExpandLifecycleHook hook = new PolicyExpandLifecycleHook(writer);
+
+ SessionUtils.doMultipleWithCommit(() -> hook.afterPersist(policy(7L, true, compaction(CRON))));
+ Assertions.assertTrue(reader.scheduledExecutionTime(7L).isPresent());
+
+ SessionUtils.doMultipleWithCommit(() -> hook.afterUpdate(policy(7L, false, compaction(CRON))));
+ Assertions.assertTrue(reader.scheduledExecutionTime(7L).isEmpty());
+ }
+
+ @TestTemplate
+ void testNonMaintenancePolicyRemovesAnyStrayRow() {
+ PolicyExpandRegistrar writer = PolicyExpandRegistrar.joiningCurrentTransaction();
+ PolicyExpandRegistrar reader = PolicyExpandRegistrar.forReads(entityStoreDataSource());
+ PolicyExpandLifecycleHook hook = new PolicyExpandLifecycleHook(writer);
+ PolicyContent custom = PolicyContents.custom(Map.of(), Set.of(MetadataObject.Type.TABLE), null);
+
+ SessionUtils.doMultipleWithCommit(() -> writer.sync(1L, true, CRON));
+ Assertions.assertTrue(reader.scheduledExecutionTime(1L).isPresent());
+
+ SessionUtils.doMultipleWithCommit(
+ () -> hook.afterPersist(policy(1L, true, custom, Policy.BuiltInType.CUSTOM)));
+ Assertions.assertTrue(reader.scheduledExecutionTime(1L).isEmpty());
+ }
+
+ @TestTemplate
+ void testDeleteRemovesByPolicyId() {
+ PolicyExpandRegistrar writer = PolicyExpandRegistrar.joiningCurrentTransaction();
+ PolicyExpandRegistrar reader = PolicyExpandRegistrar.forReads(entityStoreDataSource());
+ PolicyExpandLifecycleHook hook = new PolicyExpandLifecycleHook(writer);
+
+ SessionUtils.doMultipleWithCommit(() -> writer.sync(11L, true, CRON));
+ Assertions.assertTrue(reader.scheduledExecutionTime(11L).isPresent());
+
+ SessionUtils.doMultipleWithCommit(() -> hook.beforeRemove(11L));
+ Assertions.assertTrue(reader.scheduledExecutionTime(11L).isEmpty());
+ }
+
+ private static DataSource entityStoreDataSource() {
+ return SqlSessionFactoryHelper.getInstance()
+ .getSqlSessionFactory()
+ .getConfiguration()
+ .getEnvironment()
+ .getDataSource();
+ }
+
+ private static PolicyEntity policy(long id, boolean enabled, PolicyContent content) {
+ return policy(id, enabled, content, Policy.BuiltInType.ICEBERG_COMPACTION);
+ }
+
+ private static PolicyEntity policy(
+ long id, boolean enabled, PolicyContent content, Policy.BuiltInType type) {
+ return PolicyEntity.builder()
+ .withId(id)
+ .withName("nightly")
+ .withNamespace(NamespaceUtil.ofPolicy("ml"))
+ .withPolicyType(type)
+ .withContent(content)
+ .withEnabled(enabled)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build();
+ }
+
+ private static PolicyContent compaction(String timeTrigger) {
+ return PolicyContents.icebergDataCompaction(
+ IcebergDataCompactionContent.DEFAULT_MIN_DATA_FILE_MSE,
+ IcebergDataCompactionContent.DEFAULT_MIN_DELETE_FILE_NUMBER,
+ IcebergDataCompactionContent.DEFAULT_DATA_FILE_MSE_WEIGHT,
+ IcebergDataCompactionContent.DEFAULT_DELETE_FILE_NUMBER_WEIGHT,
+ IcebergDataCompactionContent.DEFAULT_MAX_PARTITION_NUM,
+ IcebergDataCompactionContent.DEFAULT_REWRITE_STRATEGY,
+ IcebergDataCompactionContent.DEFAULT_SORT_ORDER,
+ IcebergDataCompactionContent.DEFAULT_REWRITE_OPTIONS,
+ timeTrigger,
+ null,
+ null,
+ null,
+ null);
+ }
+}
diff --git a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/policy/TestPolicyExpandReconciler.java b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/policy/TestPolicyExpandReconciler.java
new file mode 100644
index 00000000000..a2931aafe17
--- /dev/null
+++ b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/policy/TestPolicyExpandReconciler.java
@@ -0,0 +1,249 @@
+/*
+ * 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 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.maintenance.policy;
+
+import com.github.kagkarlsson.scheduler.task.ExecutionComplete;
+import com.github.kagkarlsson.scheduler.task.ExecutionOperations;
+import java.io.IOException;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.List;
+import javax.sql.DataSource;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.BaseMetalake;
+import org.apache.gravitino.meta.PolicyEntity;
+import org.apache.gravitino.policy.IcebergDataCompactionContent;
+import org.apache.gravitino.policy.Policy;
+import org.apache.gravitino.policy.PolicyContent;
+import org.apache.gravitino.policy.PolicyContents;
+import org.apache.gravitino.storage.RandomIdGenerator;
+import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.apache.gravitino.storage.relational.service.MetalakeMetaService;
+import org.apache.gravitino.storage.relational.service.PolicyMetaService;
+import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
+import org.apache.gravitino.utils.NamespaceUtil;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.TestTemplate;
+import org.mockito.Mockito;
+
+class TestPolicyExpandReconciler extends TestJDBCBackend {
+
+ private static final String CRON = "0 0 2 * * ?";
+ private static final String LATER_CRON = "0 30 4 * * ?";
+ private static final String METALAKE = "metalake_for_expand_reconcile";
+
+ @TestTemplate
+ void testReconcileBackfillsExpandRowForExistingPolicy() throws IOException {
+ createAndInsertMakeLake(METALAKE);
+
+ PolicyEntity policy =
+ PolicyEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("nightly")
+ .withNamespace(NamespaceUtil.ofPolicy(METALAKE))
+ .withPolicyType(Policy.BuiltInType.ICEBERG_COMPACTION)
+ .withContent(compaction(CRON))
+ .withEnabled(true)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build();
+ PolicyMetaService.getInstance().insertPolicy(policy, false);
+
+ DataSource dataSource = entityStoreDataSource();
+ PolicyExpandRegistrar reader = PolicyExpandRegistrar.forReads(dataSource);
+ Assertions.assertTrue(reader.scheduledExecutionTime(policy.id()).isEmpty());
+
+ PolicyExpandRegistrar writer = PolicyExpandRegistrar.joiningCurrentTransaction();
+ PolicyExpandLifecycleHook aligner = new PolicyExpandLifecycleHook(writer);
+ int aligned =
+ new PolicyExpandReconciler(
+ aligner, writer, MetalakeMetaService.getInstance(), PolicyMetaService.getInstance())
+ .reconcileAll();
+
+ Assertions.assertTrue(aligned >= 1);
+ Assertions.assertTrue(reader.scheduledExecutionTime(policy.id()).isPresent());
+ }
+
+ @TestTemplate
+ void testReconcilePreservesOverdueExpandDue() throws IOException {
+ createAndInsertMakeLake(METALAKE + "_overdue");
+ String metalake = METALAKE + "_overdue";
+ PolicyEntity policy =
+ PolicyEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("overdue")
+ .withNamespace(NamespaceUtil.ofPolicy(metalake))
+ .withPolicyType(Policy.BuiltInType.ICEBERG_COMPACTION)
+ .withContent(compaction(CRON))
+ .withEnabled(true)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build();
+ PolicyMetaService.getInstance().insertPolicy(policy, false);
+
+ // db-scheduler / JDBC timestamps are microsecond precision; truncate so round-trip equals.
+ Instant overdue = Instant.now().minusSeconds(3600).truncatedTo(ChronoUnit.MICROS);
+ PolicyExpandRegistrar writer = PolicyExpandRegistrar.joiningCurrentTransaction();
+ PolicyExpandRegistrar reader = PolicyExpandRegistrar.forReads(entityStoreDataSource());
+ SessionUtils.doMultipleWithCommit(() -> writer.scheduleAt(policy.id(), overdue));
+ Assertions.assertEquals(overdue, reader.scheduledExecutionTime(policy.id()).orElseThrow());
+
+ PolicyExpandLifecycleHook aligner = new PolicyExpandLifecycleHook(writer);
+ new PolicyExpandReconciler(
+ aligner, writer, MetalakeMetaService.getInstance(), PolicyMetaService.getInstance())
+ .reconcileAll();
+
+ Assertions.assertEquals(overdue, reader.scheduledExecutionTime(policy.id()).orElseThrow());
+ }
+
+ @TestTemplate
+ void testReconcileCorrectsStaleExpandDueTime() throws IOException {
+ createAndInsertMakeLake(METALAKE + "_stale");
+ String metalake = METALAKE + "_stale";
+ PolicyEntity policy =
+ PolicyEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("stale-due")
+ .withNamespace(NamespaceUtil.ofPolicy(metalake))
+ .withPolicyType(Policy.BuiltInType.ICEBERG_COMPACTION)
+ .withContent(compaction(CRON))
+ .withEnabled(true)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build();
+ PolicyMetaService.getInstance().insertPolicy(policy, false);
+
+ PolicyExpandRegistrar writer = PolicyExpandRegistrar.joiningCurrentTransaction();
+ PolicyExpandRegistrar reader = PolicyExpandRegistrar.forReads(entityStoreDataSource());
+ SessionUtils.doMultipleWithCommit(() -> writer.sync(policy.id(), true, LATER_CRON));
+ Instant staleDue = reader.scheduledExecutionTime(policy.id()).orElseThrow();
+
+ PolicyExpandLifecycleHook aligner = new PolicyExpandLifecycleHook(writer);
+ new PolicyExpandReconciler(
+ aligner, writer, MetalakeMetaService.getInstance(), PolicyMetaService.getInstance())
+ .reconcileAll();
+
+ Instant corrected = reader.scheduledExecutionTime(policy.id()).orElseThrow();
+ Assertions.assertNotEquals(staleDue, corrected);
+ }
+
+ @TestTemplate
+ void testReconcileRemovesExpandRowForInvalidCron() {
+ PolicyEntity bad =
+ PolicyEntity.builder()
+ .withId(7L)
+ .withName("bad")
+ .withNamespace(NamespaceUtil.ofPolicy("ml"))
+ .withPolicyType(Policy.BuiltInType.ICEBERG_COMPACTION)
+ .withContent(compaction("not a cron"))
+ .withEnabled(true)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build();
+ BaseMetalake metalake = Mockito.mock(BaseMetalake.class);
+ Mockito.when(metalake.name()).thenReturn("ml");
+
+ MetalakeMetaService metalakes = Mockito.mock(MetalakeMetaService.class);
+ PolicyMetaService policies = Mockito.mock(PolicyMetaService.class);
+ Mockito.when(metalakes.listMetalakes()).thenReturn(List.of(metalake));
+ Mockito.when(policies.listPoliciesByNamespace(NamespaceUtil.ofPolicy("ml")))
+ .thenReturn(List.of(bad));
+
+ PolicyExpandRegistrar writer = PolicyExpandRegistrar.joiningCurrentTransaction();
+ SessionUtils.doMultipleWithCommit(() -> writer.sync(7L, true, CRON));
+ PolicyExpandRegistrar reader = PolicyExpandRegistrar.forReads(entityStoreDataSource());
+ Assertions.assertTrue(reader.scheduledExecutionTime(7L).isPresent());
+
+ PolicyExpandLifecycleHook aligner = new PolicyExpandLifecycleHook(writer);
+ Assertions.assertDoesNotThrow(
+ () -> new PolicyExpandReconciler(aligner, writer, metalakes, policies).reconcileAll());
+ Assertions.assertTrue(reader.scheduledExecutionTime(7L).isEmpty());
+ }
+
+ @TestTemplate
+ void testReconcileRemovesOrphanExpandRow() {
+ PolicyExpandRegistrar writer = PolicyExpandRegistrar.joiningCurrentTransaction();
+ SessionUtils.doMultipleWithCommit(() -> writer.sync(99L, true, CRON));
+ PolicyExpandRegistrar reader = PolicyExpandRegistrar.forReads(entityStoreDataSource());
+ Assertions.assertTrue(reader.scheduledExecutionTime(99L).isPresent());
+
+ MetalakeMetaService metalakes = Mockito.mock(MetalakeMetaService.class);
+ PolicyMetaService policies = Mockito.mock(PolicyMetaService.class);
+ Mockito.when(metalakes.listMetalakes()).thenReturn(List.of());
+
+ PolicyExpandLifecycleHook aligner = new PolicyExpandLifecycleHook(writer);
+ new PolicyExpandReconciler(aligner, writer, metalakes, policies).reconcileAll();
+ Assertions.assertTrue(reader.scheduledExecutionTime(99L).isEmpty());
+ }
+
+ @TestTemplate
+ void testCompletePickReschedulesEnabledPolicy() throws IOException {
+ createAndInsertMakeLake(METALAKE + "_pick");
+ String metalake = METALAKE + "_pick";
+ PolicyEntity policy =
+ PolicyEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("picked")
+ .withNamespace(NamespaceUtil.ofPolicy(metalake))
+ .withPolicyType(Policy.BuiltInType.ICEBERG_COMPACTION)
+ .withContent(compaction(CRON))
+ .withEnabled(true)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build();
+ PolicyMetaService.getInstance().insertPolicy(policy, false);
+
+ ExecutionOperations
+ *
+ */
+public final class PolicyExpandReconciler {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PolicyExpandReconciler.class);
+
+ private final PolicyExpandLifecycleHook aligner;
+ private final PolicyExpandRegistrar registrar;
+ private final MetalakeMetaService metalakeMetaService;
+ private final PolicyMetaService policyMetaService;
+
+ /**
+ * @param aligner same expand rules as the lifecycle hook
+ * @param registrar writer used for orphan cleanup; both must use {@link
+ * PolicyExpandRegistrar#joiningCurrentTransaction()}
+ */
+ public PolicyExpandReconciler(
+ PolicyExpandLifecycleHook aligner, PolicyExpandRegistrar registrar) {
+ this(aligner, registrar, MetalakeMetaService.getInstance(), PolicyMetaService.getInstance());
+ }
+
+ /**
+ * @param aligner expand aligner
+ * @param registrar expand registrar for orphan deletes
+ * @param metalakeMetaService metalake listing
+ * @param policyMetaService policy listing
+ */
+ public PolicyExpandReconciler(
+ PolicyExpandLifecycleHook aligner,
+ PolicyExpandRegistrar registrar,
+ MetalakeMetaService metalakeMetaService,
+ PolicyMetaService policyMetaService) {
+ this.aligner = aligner;
+ this.registrar = registrar;
+ this.metalakeMetaService = metalakeMetaService;
+ this.policyMetaService = policyMetaService;
+ }
+
+ /**
+ * For each policy, insert or correct its expand task; then delete orphan expand tasks.
+ *
+ * @return number of policies walked
+ */
+ public int reconcileAll() {
+ Set