This is an automated email from the ASF dual-hosted git repository. roryqi pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit 0763327dc8c61277ef8be68edc3456601f9ddb86 Author: remilu <[email protected]> AuthorDate: Mon Oct 26 07:55:09 2026 +0800 [#14218] feat(tms): Commit-chain collection and auth consistency Fixes #14218. Squashed from fork #79 (B9). --- .../scheduler/job/commit/CollectedCommitChain.java | 56 ++++++ .../job/commit/CommitAuthConsistency.java | 179 +++++++++++++++++++ .../scheduler/job/commit/CommitChainCollector.java | 195 +++++++++++++++++++++ .../scheduler/job/commit/CommitChainMember.java | 64 +++++++ .../job/commit/CommitJobUpsertResult.java | 40 +++++ .../scheduler/job/commit/CommitOccupancyGate.java | 65 +++++++ .../maintenance/storage/MetalakeNames.java | 54 ++++++ .../job/commit/TestCommitAuthConsistency.java | 145 +++++++++++++++ 8 files changed, 798 insertions(+) diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CollectedCommitChain.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CollectedCommitChain.java new file mode 100644 index 0000000000..b37cf9717f --- /dev/null +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CollectedCommitChain.java @@ -0,0 +1,56 @@ +/* + * 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.scheduler.job.commit; + +import com.google.common.base.Preconditions; +import java.util.List; +import org.apache.gravitino.maintenance.scheduler.job.cron.MaintenanceTableRefs.ResolvedTableRef; + +/** + * Result of {@link CommitChainCollector#collect(long)}: ordered onCommit members plus the resolved + * Spark table ref used to build the enqueue-time spec (single table-metadata lookup). + */ +public final class CollectedCommitChain { + + private final List<CommitChainMember> members; + private final ResolvedTableRef tableRef; + + /** + * @param members ordered chain members (may be empty) + * @param tableRef resolved Spark table ref for the table id + */ + public CollectedCommitChain(List<CommitChainMember> members, ResolvedTableRef tableRef) { + this.members = List.copyOf(Preconditions.checkNotNull(members, "members")); + this.tableRef = Preconditions.checkNotNull(tableRef, "tableRef"); + } + + /** + * @return ordered onCommit members + */ + public List<CommitChainMember> members() { + return members; + } + + /** + * @return resolved Spark table ref + */ + public ResolvedTableRef tableRef() { + return tableRef; + } +} diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CommitAuthConsistency.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CommitAuthConsistency.java new file mode 100644 index 0000000000..270a474148 --- /dev/null +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CommitAuthConsistency.java @@ -0,0 +1,179 @@ +/* + * 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.scheduler.job.commit; + +import com.google.common.base.Preconditions; +import java.util.ArrayList; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.TreeMap; +import java.util.TreeSet; +import javax.annotation.Nullable; +import org.apache.gravitino.Config; +import org.apache.gravitino.maintenance.policy.MaintenanceAuthConfigs; +import org.apache.gravitino.meta.PolicyEntity; +import org.apache.gravitino.policy.ExecutablePolicyContent; +import org.apache.gravitino.policy.PolicyContent; +import org.apache.gravitino.policy.PolicyExecutionSpec; + +/** + * Fail-closed checks that commit-chain members agree on effective Gravitino / IRC auth after + * cluster overlay. + * + * <p>Cluster-static {@code gravitinoAuth} / {@code ircAuth} normally make every policy identical. + * When conf leaves those bags empty, policies may carry different auth-shaped options; a single + * chained Spark Job cannot pick one identity silently. + */ +final class CommitAuthConsistency { + + private CommitAuthConsistency() {} + + /** + * @param chain filtered commit-chain members (same table) + * @param gravitinoConfig server config used for auth overlay + * @throws IllegalArgumentException when effective auth bags disagree + */ + static void requireConsistent(List<CommitChainMember> chain, Config gravitinoConfig) { + Preconditions.checkNotNull(chain); + Preconditions.checkNotNull(gravitinoConfig); + if (chain.size() <= 1) { + return; + } + + Set<String> clusterIrcKeys = MaintenanceAuthConfigs.loadIrcAuth(gravitinoConfig).keySet(); + Map<String, String> expectedGravitino = null; + Map<String, String> expectedIrc = null; + long expectedPolicyId = -1L; + + for (CommitChainMember member : chain) { + PolicyEntity policy = member.policy(); + PolicyExecutionSpec spec = executionSpec(policy.content()); + Map<String, String> gravitinoAuth = + projectGravitinoAuth( + MaintenanceAuthConfigs.overlayUpdaterOptions(gravitinoConfig, spec.updaterOptions())); + Map<String, String> ircAuth = + projectIrcAuth( + MaintenanceAuthConfigs.overlayJobOptions(gravitinoConfig, spec.jobOptions()), + clusterIrcKeys); + + if (expectedGravitino == null) { + expectedGravitino = gravitinoAuth; + expectedIrc = ircAuth; + expectedPolicyId = policy.id(); + continue; + } + if (!expectedGravitino.equals(gravitinoAuth)) { + throw new IllegalArgumentException( + mismatchMessage( + "gravitinoAuth", expectedPolicyId, policy.id(), expectedGravitino, gravitinoAuth)); + } + if (!Objects.equals(expectedIrc, ircAuth)) { + throw new IllegalArgumentException( + mismatchMessage("ircAuth", expectedPolicyId, policy.id(), expectedIrc, ircAuth)); + } + } + } + + /** + * Builds a mismatch message with policy ids and differing key names only — never auth values + * (passwords / tokens must not land in server logs). + */ + static String mismatchMessage( + String bagName, + long expectedPolicyId, + long actualPolicyId, + Map<String, String> expected, + Map<String, String> actual) { + return String.format( + "commit-chain %s mismatch: policy %s vs policy %s; differing keys=%s", + bagName, expectedPolicyId, actualPolicyId, differingKeys(expected, actual)); + } + + /** Key names that are missing, extra, or present with unequal values. */ + static List<String> differingKeys(Map<String, String> left, Map<String, String> right) { + Set<String> names = new TreeSet<>(); + names.addAll(left.keySet()); + names.addAll(right.keySet()); + List<String> differing = new ArrayList<>(); + for (String key : names) { + if (!Objects.equals(left.get(key), right.get(key))) { + differing.add(key); + } + } + return differing; + } + + static Map<String, String> projectGravitinoAuth(@Nullable Map<String, String> updaterOptions) { + Map<String, String> projected = new TreeMap<>(); + if (updaterOptions == null) { + return projected; + } + for (String key : MaintenanceAuthConfigs.GRAVITINO_AUTH_UPDATER_OPTION_KEYS) { + String value = updaterOptions.get(key); + if (value != null) { + projected.put(key, value); + } + } + return projected; + } + + /** + * Projects IRC auth-shaped keys: every cluster {@code ircAuth} key present after overlay, plus + * any remaining jobOption key that looks like catalog auth (so policy-only auth still conflicts). + */ + static Map<String, String> projectIrcAuth( + @Nullable Map<String, String> overlaidJobOptions, Set<String> clusterIrcKeys) { + Map<String, String> projected = new TreeMap<>(); + if (overlaidJobOptions == null) { + return projected; + } + for (Map.Entry<String, String> entry : overlaidJobOptions.entrySet()) { + String key = entry.getKey(); + if (key == null || entry.getValue() == null) { + continue; + } + if (clusterIrcKeys.contains(key) || isIrcAuthShaped(key)) { + projected.put(key, entry.getValue()); + } + } + return projected; + } + + static boolean isIrcAuthShaped(String key) { + String lower = key.toLowerCase(Locale.ROOT); + return lower.contains(".auth.") + || lower.startsWith("auth.") + || lower.startsWith("rest.auth") + || lower.contains("oauth") + || lower.contains("keytab") + || lower.contains("credential") + || lower.endsWith(".token") + || lower.endsWith(".password"); + } + + private static PolicyExecutionSpec executionSpec(@Nullable PolicyContent content) { + if (content instanceof ExecutablePolicyContent) { + return ((ExecutablePolicyContent) content).executionSpec(); + } + return PolicyExecutionSpec.empty(); + } +} diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CommitChainCollector.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CommitChainCollector.java new file mode 100644 index 0000000000..62a96f17af --- /dev/null +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CommitChainCollector.java @@ -0,0 +1,195 @@ +/* + * 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.scheduler.job.commit; + +import com.google.common.base.Preconditions; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Optional; +import java.util.function.Function; +import java.util.function.Supplier; +import javax.annotation.Nullable; +import org.apache.gravitino.EntityStore; +import org.apache.gravitino.GravitinoEnv; +import org.apache.gravitino.MetadataObject; +import org.apache.gravitino.MetadataObjects; +import org.apache.gravitino.maintenance.scheduler.job.cron.MaintenanceJobTemplates; +import org.apache.gravitino.maintenance.scheduler.job.cron.MaintenanceTableRefs; +import org.apache.gravitino.maintenance.scheduler.job.cron.MaintenanceTableRefs.ResolvedTableRef; +import org.apache.gravitino.maintenance.scheduler.policy.expand.NearestMaintenancePolicyResolver; +import org.apache.gravitino.maintenance.scheduler.policy.expand.NearestMaintenancePolicyResolver.Decision; +import org.apache.gravitino.maintenance.storage.MetalakeNames; +import org.apache.gravitino.meta.PolicyEntity; +import org.apache.gravitino.policy.ExecutablePolicyContent; +import org.apache.gravitino.policy.Policy; +import org.apache.gravitino.policy.PolicyContent; +import org.apache.gravitino.storage.relational.service.PolicyMetaService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Builds the onCommit maintenance chain for a table: compaction → manifest-rewrite → + * snapshot-expiry (§5.2.3). Orphan cleanup is timeTrigger-only and never included. + */ +public final class CommitChainCollector { + + private static final Logger LOG = LoggerFactory.getLogger(CommitChainCollector.class); + + /** Fixed commit-chain order. */ + public static final List<Policy.BuiltInType> CHAIN_ORDER = + List.of( + Policy.BuiltInType.ICEBERG_COMPACTION, + Policy.BuiltInType.ICEBERG_REWRITE_MANIFESTS, + Policy.BuiltInType.ICEBERG_SNAPSHOT_EXPIRATION); + + private final Function<Long, Optional<PolicyEntity>> policyLoader; + private final Supplier<EntityStore> entityStoreSupplier; + private final Function<Long, Optional<TableContext>> tableContextResolver; + + /** + * Production collector using {@link PolicyMetaService}, {@link GravitinoEnv#entityStore()}, and + * relational table refs. + */ + public CommitChainCollector() { + this( + id -> Optional.ofNullable(PolicyMetaService.getInstance().getPolicyById(id)), + () -> GravitinoEnv.getInstance().entityStore(), + CommitChainCollector::resolveTableContext); + } + + /** + * Visible for tests. + * + * @param policyLoader loads policies by id + * @param entityStoreSupplier entity store supplier + * @param tableContextResolver resolves metalake + metadata object for a table id + */ + CommitChainCollector( + Function<Long, Optional<PolicyEntity>> policyLoader, + Supplier<EntityStore> entityStoreSupplier, + Function<Long, Optional<TableContext>> tableContextResolver) { + this.policyLoader = Preconditions.checkNotNull(policyLoader); + this.entityStoreSupplier = Preconditions.checkNotNull(entityStoreSupplier); + this.tableContextResolver = Preconditions.checkNotNull(tableContextResolver); + } + + /** + * Collects the onCommit chain and the Spark table ref in one metadata resolution pass. + * + * @param tableId table entity id + * @return collected chain (members may be empty) when the table resolves; empty when metadata is + * unknown + */ + public Optional<CollectedCommitChain> collect(long tableId) { + Optional<TableContext> ctx = tableContextResolver.apply(tableId); + if (ctx.isEmpty()) { + LOG.info("Skipping commit chain for table {}: could not resolve table metadata", tableId); + return Optional.empty(); + } + TableContext table = ctx.get(); + NearestMaintenancePolicyResolver nearest = + new NearestMaintenancePolicyResolver(entityStoreSupplier.get()); + List<CommitChainMember> chain = new ArrayList<>(); + for (Policy.BuiltInType type : CHAIN_ORDER) { + Decision decision = nearest.resolve(table.metalake(), table.tableObject(), type); + if (decision.kind() == Decision.Kind.CONFLICT) { + // Match expand fail-closed: ambiguous same-level policies skip the whole table. + LOG.error( + "Skip commit-job enqueue for table {}: ambiguous {} policies {} on nearest level", + tableId, + type, + decision.conflictingPolicyIds()); + return Optional.empty(); + } + Optional<Long> winnerId = decision.winnerPolicyId(); + if (winnerId.isEmpty()) { + continue; + } + Optional<PolicyEntity> loaded = policyLoader.apply(winnerId.get()); + if (loaded.isEmpty()) { + continue; + } + PolicyEntity policy = loaded.get(); + if (!policy.enabled() || !Boolean.TRUE.equals(commitTrigger(policy.content()))) { + continue; + } + Optional<String> wire = MaintenanceJobTemplates.maintenanceType(type); + if (wire.isEmpty()) { + continue; + } + chain.add(new CommitChainMember(type, policy, wire.get())); + } + return Optional.of(new CollectedCommitChain(chain, table.tableRef())); + } + + @Nullable + private static Boolean commitTrigger(@Nullable PolicyContent content) { + if (!(content instanceof ExecutablePolicyContent)) { + return null; + } + return ((ExecutablePolicyContent) content).executionSpec().commitTrigger(); + } + + private static Optional<TableContext> resolveTableContext(long tableId) { + String metalake = MetalakeNames.forTableId(tableId); + if (metalake == null || metalake.isEmpty()) { + return Optional.empty(); + } + List<ResolvedTableRef> refs = MaintenanceTableRefs.resolve(List.of(tableId)); + if (refs.isEmpty()) { + return Optional.empty(); + } + ResolvedTableRef ref = refs.get(0); + String[] parts = ref.tableIdentifier().split("\\."); + if (parts.length < 2) { + return Optional.empty(); + } + List<String> names = new ArrayList<>(); + names.add(ref.catalogName()); + names.addAll(Arrays.asList(parts)); + MetadataObject tableObject = MetadataObjects.of(names, MetadataObject.Type.TABLE); + return Optional.of(new TableContext(metalake, tableObject, ref)); + } + + /** Metalake + metadata object + Spark ref for nearest-policy walks and spec build. */ + static final class TableContext { + private final String metalake; + private final MetadataObject tableObject; + private final ResolvedTableRef tableRef; + + TableContext(String metalake, MetadataObject tableObject, ResolvedTableRef tableRef) { + this.metalake = metalake; + this.tableObject = tableObject; + this.tableRef = tableRef; + } + + String metalake() { + return metalake; + } + + MetadataObject tableObject() { + return tableObject; + } + + ResolvedTableRef tableRef() { + return tableRef; + } + } +} diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CommitChainMember.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CommitChainMember.java new file mode 100644 index 0000000000..414ad73c3e --- /dev/null +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CommitChainMember.java @@ -0,0 +1,64 @@ +/* + * 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.scheduler.job.commit; + +import com.google.common.base.Preconditions; +import org.apache.gravitino.meta.PolicyEntity; +import org.apache.gravitino.policy.Policy; + +/** One onCommit maintenance type in the commit-job chain (§5.2.3). */ +public final class CommitChainMember { + + private final Policy.BuiltInType policyType; + private final PolicyEntity policy; + private final String maintenanceType; + + /** + * @param policyType built-in maintenance type + * @param policy nearest enabled policy with {@code commitTrigger=true} + * @param maintenanceType Spark wire name ({@code rewrite_data_files}, …) + */ + public CommitChainMember( + Policy.BuiltInType policyType, PolicyEntity policy, String maintenanceType) { + this.policyType = Preconditions.checkNotNull(policyType); + this.policy = Preconditions.checkNotNull(policy); + this.maintenanceType = Preconditions.checkNotNull(maintenanceType); + } + + /** + * @return built-in policy type + */ + public Policy.BuiltInType policyType() { + return policyType; + } + + /** + * @return policy entity + */ + public PolicyEntity policy() { + return policy; + } + + /** + * @return Spark {@code maintenance_types} wire value + */ + public String maintenanceType() { + return maintenanceType; + } +} diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CommitJobUpsertResult.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CommitJobUpsertResult.java new file mode 100644 index 0000000000..3370c88eac --- /dev/null +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CommitJobUpsertResult.java @@ -0,0 +1,40 @@ +/* + * 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.scheduler.job.commit; + +/** + * Outcome of {@link CommitJobScheduler#upsertDueNow(long)}. + * + * <p>Distinguishes coalesce ({@link #ALREADY_QUEUED}) from enqueue gate skips ({@link #SKIPPED}) so + * callers do not log "already queued" when no spec was written. + */ +public enum CommitJobUpsertResult { + + /** A new due-now {@code commit-job} row was inserted with a MaintenanceJobSpec. */ + INSERTED, + + /** A {@code commit-job} row already exists for the table (coalesced; {@code task_data} kept). */ + ALREADY_QUEUED, + + /** + * Nothing written: empty / cooled-down chain, auth failure, unresolved table, or jobConf build + * failure. + */ + SKIPPED +} diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CommitOccupancyGate.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CommitOccupancyGate.java new file mode 100644 index 0000000000..38bdb04f81 --- /dev/null +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/CommitOccupancyGate.java @@ -0,0 +1,65 @@ +/* + * 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.scheduler.job.commit; + +import org.apache.gravitino.maintenance.occupancy.OccupancyGate; + +/** + * Occupancy checks for commit-job paths. + * + * <p>{@link #withinMinInterval} gates enqueue (drops cooled-down chain members before spec build). + * {@link #isInFlight} gates pick (any chain type still in-flight aborts submit). + */ +public interface CommitOccupancyGate { + + /** + * @param tableId table entity id + * @param policyId policy entity id + * @return true when that policy still holds a soft slot + */ + boolean isInFlight(long tableId, long policyId); + + /** + * @param tableId table entity id + * @param policyId policy entity id + * @param minIntervalMs resolved cooldown + * @param nowMs current epoch millis + * @return true when still inside {@code minIntervalMs} after the last finish + */ + boolean withinMinInterval(long tableId, long policyId, long minIntervalMs, long nowMs); + + /** + * @param gate production occupancy gate + * @return adapter + */ + static CommitOccupancyGate of(OccupancyGate gate) { + return new CommitOccupancyGate() { + @Override + public boolean isInFlight(long tableId, long policyId) { + return gate.isInFlight(tableId, policyId); + } + + @Override + public boolean withinMinInterval( + long tableId, long policyId, long minIntervalMs, long nowMs) { + return gate.withinMinInterval(tableId, policyId, minIntervalMs, nowMs); + } + }; + } +} diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/MetalakeNames.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/MetalakeNames.java new file mode 100644 index 0000000000..a45a6fc71d --- /dev/null +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/MetalakeNames.java @@ -0,0 +1,54 @@ +/* + * 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.storage; + +import java.util.Collections; +import java.util.List; +import javax.annotation.Nullable; +import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; +import org.apache.gravitino.storage.relational.mapper.TableMetaMapper; +import org.apache.gravitino.storage.relational.po.MetalakePO; +import org.apache.gravitino.storage.relational.po.TablePO; +import org.apache.gravitino.storage.relational.utils.SessionUtils; + +/** Resolves metalake names for TMS entity ids. */ +public final class MetalakeNames { + + private MetalakeNames() {} + + /** + * @param tableId table entity id + * @return metalake name owning the table, or {@code null} when the table/metalake is missing + */ + @Nullable + public static String forTableId(long tableId) { + List<TablePO> tables = + SessionUtils.getWithoutCommit( + TableMetaMapper.class, + mapper -> mapper.listTablePOsByTableIds(Collections.singletonList(tableId))); + if (tables == null || tables.isEmpty()) { + return null; + } + MetalakePO metalake = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, + mapper -> mapper.selectMetalakeMetaById(tables.get(0).getMetalakeId())); + return metalake == null ? null : metalake.getMetalakeName(); + } +} diff --git a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/job/commit/TestCommitAuthConsistency.java b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/job/commit/TestCommitAuthConsistency.java new file mode 100644 index 0000000000..092ed061a6 --- /dev/null +++ b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/job/commit/TestCommitAuthConsistency.java @@ -0,0 +1,145 @@ +/* + * 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.scheduler.job.commit; + +import java.time.Instant; +import java.util.List; +import java.util.Map; +import org.apache.gravitino.Config; +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.PolicyContents; +import org.apache.gravitino.utils.NamespaceUtil; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class TestCommitAuthConsistency { + + @Test + void testAgreeingAuthPasses() { + Config config = new Config(false) {}; + CommitChainMember a = member(1L, Map.of("rest.auth.type", "none"), Map.of("auth_type", "none")); + CommitChainMember b = member(2L, Map.of("rest.auth.type", "none"), Map.of("auth_type", "none")); + Assertions.assertDoesNotThrow( + () -> CommitAuthConsistency.requireConsistent(List.of(a, b), config)); + } + + @Test + void testMismatchingGravitinoAuthFails() { + Config config = new Config(false) {}; + CommitChainMember a = member(1L, Map.of(), Map.of("auth_type", "none")); + CommitChainMember b = member(2L, Map.of(), Map.of("auth_type", "simple", "username", "u")); + IllegalArgumentException ex = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> CommitAuthConsistency.requireConsistent(List.of(a, b), config)); + String message = ex.getMessage(); + Assertions.assertTrue(message.contains("gravitinoAuth mismatch")); + Assertions.assertTrue(message.contains("policy 1 vs policy 2")); + Assertions.assertTrue(message.contains("differing keys=")); + Assertions.assertTrue(message.contains("auth_type")); + Assertions.assertTrue(message.contains("username")); + } + + @Test + void testMismatchingIrcAuthFails() { + Config config = new Config(false) {}; + CommitChainMember a = member(1L, Map.of("rest.auth.type", "none"), Map.of()); + CommitChainMember b = member(2L, Map.of("rest.auth.type", "basic"), Map.of()); + IllegalArgumentException ex = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> CommitAuthConsistency.requireConsistent(List.of(a, b), config)); + String message = ex.getMessage(); + Assertions.assertTrue(message.contains("ircAuth mismatch")); + Assertions.assertTrue(message.contains("policy 1 vs policy 2")); + Assertions.assertTrue(message.contains("differing keys=[rest.auth.type]")); + } + + @Test + void testMismatchMessageOmitsAuthValues() { + Config config = new Config(false) {}; + String secretA = "SuperSecretPassword-A"; + String secretB = "SuperSecretPassword-B"; + CommitChainMember a = + member( + 1L, + Map.of("rest.auth.type", "basic", "rest.auth.basic.password", secretA), + Map.of("auth_type", "basic", "password", secretA)); + CommitChainMember b = + member( + 2L, + Map.of("rest.auth.type", "basic", "rest.auth.basic.password", secretB), + Map.of("auth_type", "basic", "password", secretB)); + IllegalArgumentException ex = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> CommitAuthConsistency.requireConsistent(List.of(a, b), config)); + String message = ex.getMessage(); + Assertions.assertFalse(message.contains(secretA), message); + Assertions.assertFalse(message.contains(secretB), message); + Assertions.assertTrue(message.contains("differing keys="), message); + Assertions.assertTrue( + message.contains("password") || message.contains("rest.auth.basic.password"), message); + } + + @Test + void testClusterIrcAuthOverlayMakesPoliciesAgree() { + Config config = new Config(false) {}; + config.loadFromMap(Map.of("gravitino.maintenance.ircAuth.rest.auth.type", "basic"), t -> true); + // Policies disagree in content, but cluster ircAuth wins on overlay. + CommitChainMember a = member(1L, Map.of("rest.auth.type", "none"), Map.of()); + CommitChainMember b = member(2L, Map.of("rest.auth.type", "oauth"), Map.of()); + Assertions.assertDoesNotThrow( + () -> CommitAuthConsistency.requireConsistent(List.of(a, b), config)); + } + + private static CommitChainMember member( + long id, Map<String, String> jobOptions, Map<String, String> updaterOptions) { + PolicyEntity policy = + PolicyEntity.builder() + .withId(id) + .withName("p-" + id) + .withNamespace(NamespaceUtil.ofPolicy("ml")) + .withPolicyType(Policy.BuiltInType.ICEBERG_COMPACTION) + .withContent( + 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, + "binpack", + "", + Map.of(), + null, + true, + null, + jobOptions, + updaterOptions)) + .withEnabled(true) + .withAuditInfo( + AuditInfo.builder().withCreator("test").withCreateTime(Instant.EPOCH).build()) + .build(); + return new CommitChainMember( + Policy.BuiltInType.ICEBERG_COMPACTION, policy, "rewrite_data_files"); + } +}
