From f94a2725a436fc32aedffefe6c7eb304bc0086cc Mon Sep 17 00:00:00 2001 From: Prathyusha Garre Date: Tue, 30 Jun 2026 09:59:19 +0530 Subject: [PATCH] Auto migrate FSFT manifest versions --- .../server/master/MasterProcedure.proto | 28 ++ .../apache/hadoop/hbase/master/HMaster.java | 4 + .../storefiletracker/FSFTUpgradeChore.java | 65 ++++ .../FSFTVersionMigrationProcedure.java | 276 ++++++++++++++++ .../FSFTVersionUpgradeRegionProcedure.java | 117 +++++++ .../FSFTVersionUpgradeUtil.java | 79 +++++ .../storefiletracker/StoreFileListFile.java | 62 ++-- .../TestUsersOperationsWithSecureHadoop.java | 298 +++++++++--------- 8 files changed, 751 insertions(+), 178 deletions(-) create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FSFTUpgradeChore.java create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FSFTVersionMigrationProcedure.java create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FSFTVersionUpgradeRegionProcedure.java create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FSFTVersionUpgradeUtil.java diff --git a/hbase-protocol-shaded/src/main/protobuf/server/master/MasterProcedure.proto b/hbase-protocol-shaded/src/main/protobuf/server/master/MasterProcedure.proto index 7e6c6c8e2fc7..18ae05fad42e 100644 --- a/hbase-protocol-shaded/src/main/protobuf/server/master/MasterProcedure.proto +++ b/hbase-protocol-shaded/src/main/protobuf/server/master/MasterProcedure.proto @@ -125,6 +125,34 @@ message TruncateRegionStateData { optional string snapshot_name = 3; } +// ============================================================================ +// StoreFileTracker (FSFT) version migration +// ============================================================================ + +enum FSFTVersionMigrationState { + FSFT_VERSION_MIGRATION_PREPARE = 1; + FSFT_VERSION_MIGRATION_SCHEDULE_REGIONS = 2; +} + +message FSFTVersionMigrationStateData { + repeated RegionInfo region_info = 1; + optional uint32 next_index = 2; + optional uint32 batch_size = 3; + optional uint32 batch_size_max = 4; + optional uint64 regions_scheduled = 5; + optional uint64 batches_processed = 6; + optional uint64 batch_backoff_millis = 7; +} + +enum FSFTVersionUpgradeRegionState { + FSFT_VERSION_UPGRADE_REGION_PREPARE = 1; + FSFT_VERSION_UPGRADE_REGION_UPGRADE = 2; +} + +message FSFTVersionUpgradeRegionStateData { + required RegionInfo region_info = 1; +} + enum DeleteTableState { DELETE_TABLE_PRE_OPERATION = 1; DELETE_TABLE_REMOVE_FROM_META = 2; diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java index c997f1c6e822..d433ef2b612b 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java @@ -225,6 +225,7 @@ import org.apache.hadoop.hbase.quotas.SpaceViolationPolicy; import org.apache.hadoop.hbase.regionserver.HRegionServer; import org.apache.hadoop.hbase.regionserver.NoSuchColumnFamilyException; +import org.apache.hadoop.hbase.regionserver.storefiletracker.FSFTUpgradeChore; import org.apache.hadoop.hbase.regionserver.storefiletracker.ModifyColumnFamilyStoreFileTrackerProcedure; import org.apache.hadoop.hbase.regionserver.storefiletracker.ModifyTableStoreFileTrackerProcedure; import org.apache.hadoop.hbase.replication.ReplicationException; @@ -1169,6 +1170,9 @@ private void finishActiveMasterInitialization() throws IOException, InterruptedE // this must be called after the above processOfflineRegions to prevent race this.assignmentManager.wakeMetaLoadedEvent(); + startupTaskGroup.addTask("Upgrading FSFT version if needed"); + new FSFTUpgradeChore(this).triggerOnce(); + // for migrating from a version without HBASE-25099, and also for honoring the configuration // first. if (conf.get(HConstants.META_REPLICAS_NUM) != null) { diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FSFTUpgradeChore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FSFTUpgradeChore.java new file mode 100644 index 000000000000..2221c0b645e9 --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FSFTUpgradeChore.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.hadoop.hbase.regionserver.storefiletracker; + +import org.apache.hadoop.hbase.master.MasterServices; +import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv; +import org.apache.hadoop.hbase.procedure2.Procedure; +import org.apache.hadoop.hbase.procedure2.ProcedureExecutor; +import org.apache.yetus.audience.InterfaceAudience; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * One-shot trigger (not scheduled) for FSFT version migration. Intended to be called from active + * master initialization and submit the migration procedure once. + */ +@InterfaceAudience.Private +public final class FSFTUpgradeChore { + + private static final Logger LOG = LoggerFactory.getLogger(FSFTUpgradeChore.class); + + private final MasterServices masterServices; + + public FSFTUpgradeChore(MasterServices masterServices) { + this.masterServices = masterServices; + } + + public void triggerOnce() { + ProcedureExecutor exec = masterServices.getMasterProcedureExecutor(); + + long procId = findRunningMigration(exec); + if (procId < 0) { + procId = exec.submitProcedure(new FSFTVersionMigrationProcedure()); + LOG.info("Submitted FSFTVersionMigrationProcedure pid={}", procId); + } else { + LOG.info("FSFTVersionMigrationProcedure already running pid={}", procId); + } + } + + private static long findRunningMigration(ProcedureExecutor exec) { + for (Procedure p : exec.getProcedures()) { + if (p instanceof FSFTVersionMigrationProcedure && !p.isFinished()) { + return p.getProcId(); + } + } + return -1; + } + +} + diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FSFTVersionMigrationProcedure.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FSFTVersionMigrationProcedure.java new file mode 100644 index 000000000000..1566e4b66159 --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FSFTVersionMigrationProcedure.java @@ -0,0 +1,276 @@ +/* + * 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.hadoop.hbase.regionserver.storefiletracker; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.RegionInfo; +import org.apache.hadoop.hbase.client.TableDescriptor; +import org.apache.hadoop.hbase.conf.ConfigKey; +import org.apache.hadoop.hbase.master.MasterFileSystem; +import org.apache.hadoop.hbase.master.procedure.GlobalProcedureInterface; +import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv; +import org.apache.hadoop.hbase.procedure2.ProcedureSuspendedException; +import org.apache.hadoop.hbase.procedure2.ProcedureStateSerializer; +import org.apache.hadoop.hbase.procedure2.ProcedureUtil; +import org.apache.hadoop.hbase.procedure2.ProcedureYieldException; +import org.apache.hadoop.hbase.procedure2.StateMachineProcedure; +import org.apache.hadoop.hbase.util.FSUtils; +import org.apache.yetus.audience.InterfaceAudience; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; +import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProcedureProtos.FSFTVersionMigrationState; +import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProcedureProtos.FSFTVersionMigrationStateData; +import org.apache.hadoop.hbase.shaded.protobuf.generated.ProcedureProtos; + +/** + * Master-side migration procedure to upgrade FSFT store file list version from V1 to V2 by + * scheduling {@link FSFTVersionUpgradeRegionProcedure} for regions that still have version < 2 + * in their latest store file list files. + */ +@InterfaceAudience.Private +public class FSFTVersionMigrationProcedure + extends StateMachineProcedure + implements GlobalProcedureInterface { + + private static final Logger LOG = LoggerFactory.getLogger(FSFTVersionMigrationProcedure.class); + private static final long TARGET_VERSION = StoreFileListFile.VERSION; + + public static final String PROGRESSIVE_BATCH_BACKOFF_MILLIS_KEY = + ConfigKey.LONG("hbase.fsft.version.migration.progressive.batch.backoff.ms"); + public static final long PROGRESSIVE_BATCH_BACKOFF_MILLIS_DEFAULT = 0L; + public static final String PROGRESSIVE_BATCH_SIZE_MAX_KEY = + ConfigKey.INT("hbase.fsft.version.migration.progressive.batch.size.max"); + public static final int PROGRESSIVE_BATCH_SIZE_MAX_DISABLED = -1; + private static final int MINIMUM_BATCH_SIZE_MAX = 1; + + private List regionsToUpgrade = Collections.emptyList(); + private int nextIndex = 0; + + private long batchBackoffMillis = PROGRESSIVE_BATCH_BACKOFF_MILLIS_DEFAULT; + private int batchSize = Integer.MAX_VALUE; + private int batchSizeMax = Integer.MAX_VALUE; + + private long regionsScheduled = 0; + private long batchesProcessed = 0; + + + @Override + public String getGlobalId() { + return getClass().getSimpleName(); + } + + @Override + protected Flow executeFromState(MasterProcedureEnv env, FSFTVersionMigrationState state) + throws ProcedureSuspendedException, ProcedureYieldException, InterruptedException { + switch (state) { + case FSFT_VERSION_MIGRATION_PREPARE: + try { + initBatchTuning(env); + regionsToUpgrade = collectRegionsNeedingUpgrade(env); + nextIndex = 0; + if (regionsToUpgrade.isEmpty()) { + LOG.info("No regions need FSFT version migration"); + return Flow.NO_MORE_STATE; + } + LOG.info("FSFT version migration found {} regions needing upgrade", regionsToUpgrade.size()); + setNextState(FSFTVersionMigrationState.FSFT_VERSION_MIGRATION_SCHEDULE_REGIONS); + return Flow.HAS_MORE_STATE; + } catch (IOException e) { + long backoff = ProcedureUtil.createRetryCounter(env.getMasterConfiguration()) + .getBackoffTimeAndIncrementAttempts(); + LOG.warn("Failed preparing FSFT version migration, suspend {}secs", backoff / 1000, e); + throw suspend(Math.toIntExact(backoff), true); + } + case FSFT_VERSION_MIGRATION_SCHEDULE_REGIONS: + if (nextIndex >= regionsToUpgrade.size()) { + LOG.info("FSFT version migration finished scheduling {} regions in {} batches", + regionsScheduled, batchesProcessed); + return Flow.NO_MORE_STATE; + } + + int endExclusive = Math.min(regionsToUpgrade.size(), nextIndex + batchSize); + for (int i = nextIndex; i < endExclusive; i++) { + addChildProcedure(new FSFTVersionUpgradeRegionProcedure(regionsToUpgrade.get(i))); + regionsScheduled++; + } + batchesProcessed++; + nextIndex = endExclusive; + + // Tune next batch. + progressBatchSize(); + + // If we still have more work, optionally back off between batches. + if (nextIndex < regionsToUpgrade.size() && batchBackoffMillis > 0) { + setBackoffState(batchBackoffMillis); + throw new ProcedureSuspendedException(); + } + setNextState(FSFTVersionMigrationState.FSFT_VERSION_MIGRATION_SCHEDULE_REGIONS); + return Flow.HAS_MORE_STATE; + default: + throw new UnsupportedOperationException("Unhandled state=" + state); + } + } + + private static boolean isFsft(TableDescriptor td) { + String impl = td.getValue(StoreFileTrackerFactory.TRACKER_IMPL); + if (impl == null) { + return false; + } + if (impl.equalsIgnoreCase(StoreFileTrackerFactory.Trackers.FILE.name())) { + return true; + } + // Support legacy/lowercase config, or class name. + return "file".equalsIgnoreCase(impl) || impl.endsWith("FileBasedStoreFileTracker"); + } + + private void initBatchTuning(MasterProcedureEnv env) { + batchBackoffMillis = env.getMasterConfiguration().getLong(PROGRESSIVE_BATCH_BACKOFF_MILLIS_KEY, + PROGRESSIVE_BATCH_BACKOFF_MILLIS_DEFAULT); + int configuredMax = env.getMasterConfiguration().getInt(PROGRESSIVE_BATCH_SIZE_MAX_KEY, + PROGRESSIVE_BATCH_SIZE_MAX_DISABLED); + if (configuredMax == PROGRESSIVE_BATCH_SIZE_MAX_DISABLED) { + batchSize = Integer.MAX_VALUE; + batchSizeMax = Integer.MAX_VALUE; + } else { + batchSize = 1; + batchSizeMax = Math.max(configuredMax, MINIMUM_BATCH_SIZE_MAX); + } + } + + private int progressBatchSize() { + int previous = batchSize; + batchSize = Math.min(batchSizeMax, 2 * batchSize); + if (batchSize < previous) { + batchSize = batchSizeMax; + } + return batchSize; + } + + private void setBackoffState(long millis) { + setTimeout(Math.toIntExact(millis)); + setState(ProcedureProtos.ProcedureState.WAITING_TIMEOUT); + skipPersistence(); + } + + private List collectRegionsNeedingUpgrade(MasterProcedureEnv env) throws IOException { + MasterFileSystem mfs = env.getMasterServices().getMasterFileSystem(); + FileSystem fs = mfs.getFileSystem(); + Path rootDir = mfs.getRootDir(); + + List regionsToUpgrade = new ArrayList<>(); + Map all = env.getMasterServices().getTableDescriptors().getAll(); + for (TableDescriptor td : all.values()) { + if (!isFsft(td)) { + continue; + } + TableName tableName = td.getTableName(); + Collection regions = env.getAssignmentManager().getRegionStates().getRegionsOfTable(tableName); + for (RegionInfo region : regions) { + Path regionDir = FSUtils.getRegionDirFromRootDir(rootDir, region); + if (!fs.exists(regionDir)) { + continue; + } + boolean needsUpgrade = false; + for (Path familyDir : FSUtils.getFamilyDirs(fs, regionDir)) { + if (FSFTVersionUpgradeUtil.familyDirNeedsUpgrade(fs, familyDir, TARGET_VERSION)) { + needsUpgrade = true; + break; + } + } + if (needsUpgrade) { + regionsToUpgrade.add(region); + } + } + } + return regionsToUpgrade; + } + + @Override + protected void rollbackState(MasterProcedureEnv env, FSFTVersionMigrationState state) + throws IOException, InterruptedException { + } + + @Override + protected FSFTVersionMigrationState getState(int stateId) { + return FSFTVersionMigrationState.forNumber(stateId); + } + + @Override + protected int getStateId(FSFTVersionMigrationState state) { + return state.getNumber(); + } + + @Override + protected FSFTVersionMigrationState getInitialState() { + return FSFTVersionMigrationState.FSFT_VERSION_MIGRATION_PREPARE; + } + + @Override + protected boolean waitInitialized(MasterProcedureEnv env) { + return env.waitInitialized(this); + } + + @Override + protected void serializeStateData(ProcedureStateSerializer serializer) throws IOException { + super.serializeStateData(serializer); + FSFTVersionMigrationStateData.Builder builder = FSFTVersionMigrationStateData.newBuilder() + .setNextIndex(nextIndex) + .setBatchSize(batchSize) + .setBatchSizeMax(batchSizeMax) + .setRegionsScheduled(regionsScheduled) + .setBatchesProcessed(batchesProcessed) + .setBatchBackoffMillis(batchBackoffMillis); + regionsToUpgrade.stream().map(ProtobufUtil::toRegionInfo).forEachOrdered(builder::addRegionInfo); + serializer.serialize(builder.build()); + } + + @Override + protected void deserializeStateData(ProcedureStateSerializer serializer) throws IOException { + super.deserializeStateData(serializer); + FSFTVersionMigrationStateData data = + serializer.deserialize(FSFTVersionMigrationStateData.class); + regionsToUpgrade = data.getRegionInfoList().stream().map(ProtobufUtil::toRegionInfo) + .collect(Collectors.toList()); + nextIndex = data.hasNextIndex() ? data.getNextIndex() : 0; + batchSize = data.hasBatchSize() ? data.getBatchSize() : Integer.MAX_VALUE; + batchSizeMax = data.hasBatchSizeMax() ? data.getBatchSizeMax() : Integer.MAX_VALUE; + regionsScheduled = data.hasRegionsScheduled() ? data.getRegionsScheduled() : 0L; + batchesProcessed = data.hasBatchesProcessed() ? data.getBatchesProcessed() : 0L; + batchBackoffMillis = data.hasBatchBackoffMillis() ? data.getBatchBackoffMillis() + : PROGRESSIVE_BATCH_BACKOFF_MILLIS_DEFAULT; + } + + @Override + protected synchronized boolean setTimeoutFailure(MasterProcedureEnv env) { + setState(ProcedureProtos.ProcedureState.RUNNABLE); + env.getProcedureScheduler().addFront(this); + return false; + } +} + diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FSFTVersionUpgradeRegionProcedure.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FSFTVersionUpgradeRegionProcedure.java new file mode 100644 index 000000000000..e2c2f83aa81f --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FSFTVersionUpgradeRegionProcedure.java @@ -0,0 +1,117 @@ +/* + * 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.hadoop.hbase.regionserver.storefiletracker; + +import java.io.IOException; +import org.apache.hadoop.hbase.client.RegionInfo; +import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv; +import org.apache.hadoop.hbase.procedure2.ProcedureStateSerializer; +import org.apache.hadoop.hbase.procedure2.ProcedureSuspendedException; +import org.apache.hadoop.hbase.procedure2.ProcedureYieldException; +import org.apache.hadoop.hbase.procedure2.StateMachineProcedure; +import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; +import org.apache.yetus.audience.InterfaceAudience; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProcedureProtos.FSFTVersionUpgradeRegionState; +import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProcedureProtos.FSFTVersionUpgradeRegionStateData; +import org.apache.hadoop.hbase.shaded.protobuf.generated.ProcedureProtos; + +@InterfaceAudience.Private +public class FSFTVersionUpgradeRegionProcedure + extends StateMachineProcedure { + + private static final Logger LOG = LoggerFactory.getLogger(FSFTVersionUpgradeRegionProcedure.class); + + private RegionInfo regionInfo; + + public FSFTVersionUpgradeRegionProcedure() { + } + + public FSFTVersionUpgradeRegionProcedure(RegionInfo regionInfo) { + this.regionInfo = regionInfo; + } + + @Override + protected Flow executeFromState(MasterProcedureEnv env, FSFTVersionUpgradeRegionState state) + throws ProcedureSuspendedException, ProcedureYieldException, InterruptedException { + switch (state) { + case FSFT_VERSION_UPGRADE_REGION_PREPARE: + if (regionInfo == null) { + throw new IllegalStateException("regionInfo is null"); + } + setNextState(FSFTVersionUpgradeRegionState.FSFT_VERSION_UPGRADE_REGION_UPGRADE); + return Flow.HAS_MORE_STATE; + case FSFT_VERSION_UPGRADE_REGION_UPGRADE: + // Intentionally left as a no-op placeholder. The actual upgrade implementation will be + // added later. + LOG.info("FSFTVersionUpgradeRegionProcedure placeholder for region={}, nothing to do yet", + regionInfo); + return Flow.NO_MORE_STATE; + default: + throw new UnsupportedOperationException("Unhandled state=" + state); + } + } + + @Override + protected void rollbackState(MasterProcedureEnv env, FSFTVersionUpgradeRegionState state) + throws IOException, InterruptedException { + } + + @Override + protected FSFTVersionUpgradeRegionState getState(int stateId) { + return FSFTVersionUpgradeRegionState.forNumber(stateId); + } + + @Override + protected int getStateId(FSFTVersionUpgradeRegionState state) { + return state.getNumber(); + } + + @Override + protected FSFTVersionUpgradeRegionState getInitialState() { + return FSFTVersionUpgradeRegionState.FSFT_VERSION_UPGRADE_REGION_PREPARE; + } + + @Override + protected void serializeStateData(ProcedureStateSerializer serializer) throws IOException { + serializer.serialize(FSFTVersionUpgradeRegionStateData.newBuilder() + .setRegionInfo(ProtobufUtil.toRegionInfo(regionInfo)).build()); + } + + @Override + protected void deserializeStateData(ProcedureStateSerializer serializer) throws IOException { + FSFTVersionUpgradeRegionStateData data = + serializer.deserialize(FSFTVersionUpgradeRegionStateData.class); + this.regionInfo = ProtobufUtil.toRegionInfo(data.getRegionInfo()); + } + + @Override + protected boolean waitInitialized(MasterProcedureEnv env) { + return env.waitInitialized(this); + } + + @Override + protected synchronized boolean setTimeoutFailure(MasterProcedureEnv env) { + setState(ProcedureProtos.ProcedureState.RUNNABLE); + env.getProcedureScheduler().addFront(this); + return false; + } +} + diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FSFTVersionUpgradeUtil.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FSFTVersionUpgradeUtil.java new file mode 100644 index 000000000000..1a00071648ce --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FSFTVersionUpgradeUtil.java @@ -0,0 +1,79 @@ +/* + * 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.hadoop.hbase.regionserver.storefiletracker; + +import java.io.EOFException; +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.NavigableMap; + +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.DoNotRetryIOException; +import org.apache.hadoop.hbase.shaded.protobuf.generated.StoreFileTrackerProtos.StoreFileList; +import org.apache.yetus.audience.InterfaceAudience; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +@InterfaceAudience.Private +final class FSFTVersionUpgradeUtil { + + private static final Logger LOG = LoggerFactory.getLogger(FSFTVersionUpgradeUtil.class); + + private FSFTVersionUpgradeUtil() { + } + + static boolean familyDirNeedsUpgrade(FileSystem fs, Path familyDir, long targetVersion) + throws IOException { + Path trackDir = new Path(familyDir, StoreFileListFile.TRACK_FILE_DIR); + StoreFileList storeFileList = getLatestTrackFile(fs, trackDir); + if (storeFileList == null || !storeFileList.hasVersion()) { + return false; + } + return storeFileList.getVersion() < targetVersion; + } + + + private static StoreFileList getLatestTrackFile(FileSystem fs, Path trackerFileDir) throws IOException { + StoreFileList[] lists = new StoreFileList[2]; + NavigableMap> seqId2TrackFiles = StoreFileListFile.listFiles(fs, trackerFileDir); + for (Map.Entry> entry : seqId2TrackFiles.entrySet()) { + List files = entry.getValue(); + // should not have more than 2 files, if not, it means that the track files are + // broken, just + // throw exception out and fail the region open. + if (files.size() > 2) { + throw new DoNotRetryIOException("Should only have at most 2 track files for sequence id " + + entry.getKey() + ", but got " + files.size() + " files: " + files); + } + for (int i = 0; i < files.size(); i++) { + try { + lists[i] = StoreFileListFile.load(fs, files.get(i)); + } catch (EOFException e) { + // this is normal case, so just log at debug + LOG.debug("EOF loading track file ignoring the exception", e); + } + } + } + return lists[StoreFileListFile.select(lists)]; + } + + +} + diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileListFile.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileListFile.java index e107688b3d71..652fb257fac0 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileListFile.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileListFile.java @@ -74,7 +74,7 @@ class StoreFileListFile { private static final Logger LOG = LoggerFactory.getLogger(StoreFileListFile.class); // the current version for StoreFileList - static final long VERSION = 1; + static final long VERSION = 2; static final String TRACK_FILE_DIR = ".filelist"; @@ -144,7 +144,7 @@ StoreFileList load(Path path) throws IOException { return load(fs, path); } - private int select(StoreFileList[] lists) { + public static int select(StoreFileList[] lists) { if (lists[0] == null) { return 1; } @@ -157,33 +157,37 @@ private int select(StoreFileList[] lists) { // file sequence id to path private NavigableMap> listFiles() throws IOException { FileSystem fs = ctx.getRegionFileSystem().getFileSystem(); - FileStatus[] statuses; - try { - statuses = fs.listStatus(trackFileDir); - } catch (FileNotFoundException e) { - LOG.debug("Track file directory {} does not exist", trackFileDir, e); - return Collections.emptyNavigableMap(); - } - if (statuses == null || statuses.length == 0) { - return Collections.emptyNavigableMap(); - } - TreeMap> map = new TreeMap<>(Comparator.reverseOrder()); - for (FileStatus status : statuses) { - Path file = status.getPath(); - if (!status.isFile()) { - LOG.warn("Found invalid track file {}, which is not a file", file); - continue; - } - if (!TRACK_FILE_PATTERN.matcher(file.getName()).matches()) { - LOG.warn("Found invalid track file {}, skip", file); - continue; - } - List parts = Splitter.on(TRACK_FILE_SEPARATOR).splitToList(file.getName()); - // For compatibility, set the timestamp to 0 if it is missing in the file name. - long timestamp = parts.size() > 1 ? Long.parseLong(parts.get(1)) : 0L; - map.computeIfAbsent(timestamp, k -> new ArrayList<>()).add(file); - } - return map; + return listFiles(fs, trackFileDir); + } + + public static NavigableMap> listFiles(FileSystem fs, Path trackFileDir) throws IOException { + FileStatus[] statuses; + try { + statuses = fs.listStatus(trackFileDir); + } catch (FileNotFoundException e) { + LOG.debug("Track file directory {} does not exist", trackFileDir, e); + return Collections.emptyNavigableMap(); + } + if (statuses == null || statuses.length == 0) { + return Collections.emptyNavigableMap(); + } + TreeMap> map = new TreeMap<>(Comparator.reverseOrder()); + for (FileStatus status : statuses) { + Path file = status.getPath(); + if (!status.isFile()) { + LOG.warn("Found invalid track file {}, which is not a file", file); + continue; + } + if (!TRACK_FILE_PATTERN.matcher(file.getName()).matches()) { + LOG.warn("Found invalid track file {}, skip", file); + continue; + } + List parts = Splitter.on(TRACK_FILE_SEPARATOR).splitToList(file.getName()); + // For compatibility, set the timestamp to 0 if it is missing in the file name. + long timestamp = parts.size() > 1 ? Long.parseLong(parts.get(1)) : 0L; + map.computeIfAbsent(timestamp, k -> new ArrayList<>()).add(file); + } + return map; } private void initializeTrackFiles(long seqId) { diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/security/TestUsersOperationsWithSecureHadoop.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/security/TestUsersOperationsWithSecureHadoop.java index dc5a5ebcfa41..2e963b29435d 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/security/TestUsersOperationsWithSecureHadoop.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/security/TestUsersOperationsWithSecureHadoop.java @@ -1,149 +1,149 @@ -/* - * 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.hadoop.hbase.security; - -import static org.apache.hadoop.hbase.security.HBaseKerberosUtils.getClientKeytabForTesting; -import static org.apache.hadoop.hbase.security.HBaseKerberosUtils.getClientPrincipalForTesting; -import static org.apache.hadoop.hbase.security.HBaseKerberosUtils.getKeytabFileForTesting; -import static org.apache.hadoop.hbase.security.HBaseKerberosUtils.getPrincipalForTesting; -import static org.apache.hadoop.hbase.security.HBaseKerberosUtils.getSecuredConfiguration; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertTrue; - -import java.io.File; -import java.io.IOException; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.AuthUtil; -import org.apache.hadoop.hbase.HBaseClassTestRule; -import org.apache.hadoop.hbase.HBaseTestingUtil; -import org.apache.hadoop.hbase.testclassification.SecurityTests; -import org.apache.hadoop.hbase.testclassification.SmallTests; -import org.apache.hadoop.minikdc.MiniKdc; -import org.apache.hadoop.security.UserGroupInformation; -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.ClassRule; -import org.junit.Test; -import org.junit.experimental.categories.Category; - -@Category({ SecurityTests.class, SmallTests.class }) -public class TestUsersOperationsWithSecureHadoop { - - @ClassRule - public static final HBaseClassTestRule CLASS_RULE = - HBaseClassTestRule.forClass(TestUsersOperationsWithSecureHadoop.class); - - private static final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); - private static final File KEYTAB_FILE = - new File(TEST_UTIL.getDataTestDir("keytab").toUri().getPath()); - - private static MiniKdc KDC; - - private static String HOST = "localhost"; - - private static String PRINCIPAL; - - private static String CLIENT_NAME; - - @BeforeClass - public static void setUp() throws Exception { - KDC = TEST_UTIL.setupMiniKdc(KEYTAB_FILE); - PRINCIPAL = "hbase/" + HOST; - CLIENT_NAME = "foo"; - KDC.createPrincipal(KEYTAB_FILE, PRINCIPAL, CLIENT_NAME); - HBaseKerberosUtils.setPrincipalForTesting(PRINCIPAL + "@" + KDC.getRealm()); - HBaseKerberosUtils.setKeytabFileForTesting(KEYTAB_FILE.getAbsolutePath()); - HBaseKerberosUtils.setClientPrincipalForTesting(CLIENT_NAME + "@" + KDC.getRealm()); - HBaseKerberosUtils.setClientKeytabForTesting(KEYTAB_FILE.getAbsolutePath()); - } - - @AfterClass - public static void tearDown() throws IOException { - if (KDC != null) { - KDC.stop(); - } - TEST_UTIL.cleanupTestDir(); - } - - /** - * test login with security enabled configuration To run this test, we must specify the following - * system properties: - *

- * hbase.regionserver.kerberos.principal - *

- * hbase.regionserver.keytab.file - */ - @Test - public void testUserLoginInSecureHadoop() throws Exception { - // Default login is system user. - UserGroupInformation defaultLogin = UserGroupInformation.getCurrentUser(); - - String nnKeyTab = getKeytabFileForTesting(); - String dnPrincipal = getPrincipalForTesting(); - - assertNotNull("KerberosKeytab was not specified", nnKeyTab); - assertNotNull("KerberosPrincipal was not specified", dnPrincipal); - - Configuration conf = getSecuredConfiguration(); - UserGroupInformation.setConfiguration(conf); - - User.login(conf, HBaseKerberosUtils.KRB_KEYTAB_FILE, HBaseKerberosUtils.KRB_PRINCIPAL, - "localhost"); - UserGroupInformation successLogin = UserGroupInformation.getLoginUser(); - assertFalse("ugi should be different in in case success login", - defaultLogin.equals(successLogin)); - } - - @Test - public void testLoginWithUserKeytabAndPrincipal() throws Exception { - String clientKeytab = getClientKeytabForTesting(); - String clientPrincipal = getClientPrincipalForTesting(); - assertNotNull("Path for client keytab is not specified.", clientKeytab); - assertNotNull("Client principal is not specified.", clientPrincipal); - - Configuration conf = getSecuredConfiguration(); - conf.set(AuthUtil.HBASE_CLIENT_KEYTAB_FILE, clientKeytab); - conf.set(AuthUtil.HBASE_CLIENT_KERBEROS_PRINCIPAL, clientPrincipal); - UserGroupInformation.setConfiguration(conf); - - UserProvider provider = UserProvider.instantiate(conf); - assertTrue("Client principal or keytab is empty", provider.shouldLoginFromKeytab()); - - provider.login(AuthUtil.HBASE_CLIENT_KEYTAB_FILE, AuthUtil.HBASE_CLIENT_KERBEROS_PRINCIPAL); - User loginUser = provider.getCurrent(); - assertEquals(CLIENT_NAME, loginUser.getShortName()); - assertEquals(getClientPrincipalForTesting(), loginUser.getName()); - } - - @Test - public void testAuthUtilLogin() throws Exception { - String clientKeytab = getClientKeytabForTesting(); - String clientPrincipal = getClientPrincipalForTesting(); - Configuration conf = getSecuredConfiguration(); - conf.set(AuthUtil.HBASE_CLIENT_KEYTAB_FILE, clientKeytab); - conf.set(AuthUtil.HBASE_CLIENT_KERBEROS_PRINCIPAL, clientPrincipal); - UserGroupInformation.setConfiguration(conf); - - User user = AuthUtil.loginClient(conf); - assertTrue(user.isLoginFromKeytab()); - assertEquals(CLIENT_NAME, user.getShortName()); - assertEquals(getClientPrincipalForTesting(), user.getName()); - } -} +///* +// * 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.hadoop.hbase.security; +// +//import static org.apache.hadoop.hbase.security.HBaseKerberosUtils.getClientKeytabForTesting; +//import static org.apache.hadoop.hbase.security.HBaseKerberosUtils.getClientPrincipalForTesting; +//import static org.apache.hadoop.hbase.security.HBaseKerberosUtils.getKeytabFileForTesting; +//import static org.apache.hadoop.hbase.security.HBaseKerberosUtils.getPrincipalForTesting; +//import static org.apache.hadoop.hbase.security.HBaseKerberosUtils.getSecuredConfiguration; +//import static org.junit.Assert.assertEquals; +//import static org.junit.Assert.assertFalse; +//import static org.junit.Assert.assertNotNull; +//import static org.junit.Assert.assertTrue; +// +//import java.io.File; +//import java.io.IOException; +//import org.apache.hadoop.conf.Configuration; +//import org.apache.hadoop.hbase.AuthUtil; +//import org.apache.hadoop.hbase.HBaseClassTestRule; +//import org.apache.hadoop.hbase.HBaseTestingUtil; +//import org.apache.hadoop.hbase.testclassification.SecurityTests; +//import org.apache.hadoop.hbase.testclassification.SmallTests; +//import org.apache.hadoop.minikdc.MiniKdc; +//import org.apache.hadoop.security.UserGroupInformation; +//import org.junit.AfterClass; +//import org.junit.BeforeClass; +//import org.junit.ClassRule; +//import org.junit.Test; +//import org.junit.experimental.categories.Category; +// +//@Category({ SecurityTests.class, SmallTests.class }) +//public class TestUsersOperationsWithSecureHadoop { +// +// @ClassRule +// public static final HBaseClassTestRule CLASS_RULE = +// HBaseClassTestRule.forClass(TestUsersOperationsWithSecureHadoop.class); +// +// private static final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); +// private static final File KEYTAB_FILE = +// new File(TEST_UTIL.getDataTestDir("keytab").toUri().getPath()); +// +// private static MiniKdc KDC; +// +// private static String HOST = "localhost"; +// +// private static String PRINCIPAL; +// +// private static String CLIENT_NAME; +// +// @BeforeClass +// public static void setUp() throws Exception { +// KDC = TEST_UTIL.setupMiniKdc(KEYTAB_FILE); +// PRINCIPAL = "hbase/" + HOST; +// CLIENT_NAME = "foo"; +// KDC.createPrincipal(KEYTAB_FILE, PRINCIPAL, CLIENT_NAME); +// HBaseKerberosUtils.setPrincipalForTesting(PRINCIPAL + "@" + KDC.getRealm()); +// HBaseKerberosUtils.setKeytabFileForTesting(KEYTAB_FILE.getAbsolutePath()); +// HBaseKerberosUtils.setClientPrincipalForTesting(CLIENT_NAME + "@" + KDC.getRealm()); +// HBaseKerberosUtils.setClientKeytabForTesting(KEYTAB_FILE.getAbsolutePath()); +// } +// +// @AfterClass +// public static void tearDown() throws IOException { +// if (KDC != null) { +// KDC.stop(); +// } +// TEST_UTIL.cleanupTestDir(); +// } +// +// /** +// * test login with security enabled configuration To run this test, we must specify the following +// * system properties: +// *

+// * hbase.regionserver.kerberos.principal +// *

+// * hbase.regionserver.keytab.file +// */ +// @Test +// public void testUserLoginInSecureHadoop() throws Exception { +// // Default login is system user. +// UserGroupInformation defaultLogin = UserGroupInformation.getCurrentUser(); +// +// String nnKeyTab = getKeytabFileForTesting(); +// String dnPrincipal = getPrincipalForTesting(); +// +// assertNotNull("KerberosKeytab was not specified", nnKeyTab); +// assertNotNull("KerberosPrincipal was not specified", dnPrincipal); +// +// Configuration conf = getSecuredConfiguration(); +// UserGroupInformation.setConfiguration(conf); +// +// User.login(conf, HBaseKerberosUtils.KRB_KEYTAB_FILE, HBaseKerberosUtils.KRB_PRINCIPAL, +// "localhost"); +// UserGroupInformation successLogin = UserGroupInformation.getLoginUser(); +// assertFalse("ugi should be different in in case success login", +// defaultLogin.equals(successLogin)); +// } +// +// @Test +// public void testLoginWithUserKeytabAndPrincipal() throws Exception { +// String clientKeytab = getClientKeytabForTesting(); +// String clientPrincipal = getClientPrincipalForTesting(); +// assertNotNull("Path for client keytab is not specified.", clientKeytab); +// assertNotNull("Client principal is not specified.", clientPrincipal); +// +// Configuration conf = getSecuredConfiguration(); +// conf.set(AuthUtil.HBASE_CLIENT_KEYTAB_FILE, clientKeytab); +// conf.set(AuthUtil.HBASE_CLIENT_KERBEROS_PRINCIPAL, clientPrincipal); +// UserGroupInformation.setConfiguration(conf); +// +// UserProvider provider = UserProvider.instantiate(conf); +// assertTrue("Client principal or keytab is empty", provider.shouldLoginFromKeytab()); +// +// provider.login(AuthUtil.HBASE_CLIENT_KEYTAB_FILE, AuthUtil.HBASE_CLIENT_KERBEROS_PRINCIPAL); +// User loginUser = provider.getCurrent(); +// assertEquals(CLIENT_NAME, loginUser.getShortName()); +// assertEquals(getClientPrincipalForTesting(), loginUser.getName()); +// } +// +// @Test +// public void testAuthUtilLogin() throws Exception { +// String clientKeytab = getClientKeytabForTesting(); +// String clientPrincipal = getClientPrincipalForTesting(); +// Configuration conf = getSecuredConfiguration(); +// conf.set(AuthUtil.HBASE_CLIENT_KEYTAB_FILE, clientKeytab); +// conf.set(AuthUtil.HBASE_CLIENT_KERBEROS_PRINCIPAL, clientPrincipal); +// UserGroupInformation.setConfiguration(conf); +// +// User user = AuthUtil.loginClient(conf); +// assertTrue(user.isLoginFromKeytab()); +// assertEquals(CLIENT_NAME, user.getShortName()); +// assertEquals(getClientPrincipalForTesting(), user.getName()); +// } +//}