You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by ct...@apache.org on 2013/09/07 05:28:17 UTC
[14/59] [abbrv] git commit: ACCUMULO-658 Move master to its own module
ACCUMULO-658 Move master to its own module
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/88079cc3
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/88079cc3
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/88079cc3
Branch: refs/heads/ACCUMULO-210
Commit: 88079cc3901a71371a9c2823a98348a0616a8e39
Parents: e053356
Author: Christopher Tubbs <ct...@apache.org>
Authored: Mon Aug 5 15:05:15 2013 -0400
Committer: Christopher Tubbs <ct...@apache.org>
Committed: Fri Sep 6 23:18:51 2013 -0400
----------------------------------------------------------------------
.../minicluster/MiniAccumuloCluster.java | 4 +-
pom.xml | 10 +
.../accumulo/master/state/CurrentState.java | 30 +
.../accumulo/master/state/DeadServerList.java | 80 +
.../apache/accumulo/master/state/MergeInfo.java | 107 ++
.../accumulo/master/state/MergeState.java | 49 +
.../master/state/MetaDataTableScanner.java | 196 ++
.../accumulo/master/state/TServerInstance.java | 127 ++
.../master/state/TabletLocationState.java | 97 +
.../master/state/TabletServerState.java | 76 +
.../accumulo/master/state/TabletState.java | 21 +
.../master/state/TabletStateChangeIterator.java | 188 ++
.../server/data/ServerColumnUpdate.java | 36 +
.../accumulo/server/data/ServerMutation.java | 81 +
.../server/master/state/CurrentState.java | 30 -
.../server/master/state/DeadServerList.java | 80 -
.../accumulo/server/master/state/MergeInfo.java | 107 --
.../server/master/state/MergeState.java | 49 -
.../master/state/MetaDataTableScanner.java | 196 --
.../server/master/state/TServerInstance.java | 127 --
.../master/state/TabletLocationState.java | 97 -
.../server/master/state/TabletServerState.java | 76 -
.../server/master/state/TabletState.java | 21 -
.../master/state/TabletStateChangeIterator.java | 188 --
.../server/tabletserver/TabletTime.java | 224 +++
.../tabletserver/UniqueNameAllocator.java | 79 +
.../accumulo/server/util/TableInfoUtil.java | 88 +
.../accumulo/server/util/TabletOperations.java | 78 +
server/master/pom.xml | 130 ++
.../accumulo/master/EventCoordinator.java | 63 +
.../apache/accumulo/master/LiveTServerSet.java | 398 ++++
.../java/org/apache/accumulo/master/Master.java | 1769 +++++++++++++++++
.../accumulo/master/TServerLockWatcher.java | 43 +
.../accumulo/master/TabletGroupWatcher.java | 652 +++++++
.../master/balancer/ChaoticLoadBalancer.java | 143 ++
.../master/balancer/DefaultLoadBalancer.java | 318 ++++
.../master/balancer/TableLoadBalancer.java | 147 ++
.../master/balancer/TabletBalancer.java | 151 ++
.../master/recovery/HadoopLogCloser.java | 61 +
.../accumulo/master/recovery/LogCloser.java | 27 +
.../accumulo/master/recovery/MapRLogCloser.java | 47 +
.../master/recovery/RecoveryManager.java | 177 ++
.../accumulo/master/state/Assignment.java | 30 +
.../accumulo/master/state/DistributedStore.java | 34 +
.../master/state/DistributedStoreException.java | 34 +
.../accumulo/master/state/MergeStats.java | 260 +++
.../master/state/MetaDataStateStore.java | 160 ++
.../master/state/RootTabletStateStore.java | 48 +
.../accumulo/master/state/SetGoalState.java | 48 +
.../accumulo/master/state/TableCounts.java | 39 +
.../accumulo/master/state/TableStats.java | 71 +
.../accumulo/master/state/TabletMigration.java | 36 +
.../accumulo/master/state/TabletStateStore.java | 94 +
.../apache/accumulo/master/state/ZooStore.java | 96 +
.../master/state/ZooTabletStateStore.java | 176 ++
.../accumulo/master/tableOps/BulkImport.java | 621 ++++++
.../master/tableOps/CancelCompactions.java | 99 +
.../master/tableOps/ChangeTableState.java | 64 +
.../accumulo/master/tableOps/CloneTable.java | 243 +++
.../accumulo/master/tableOps/CompactRange.java | 386 ++++
.../accumulo/master/tableOps/CreateTable.java | 288 +++
.../accumulo/master/tableOps/DeleteTable.java | 242 +++
.../accumulo/master/tableOps/ExportTable.java | 309 +++
.../accumulo/master/tableOps/ImportTable.java | 601 ++++++
.../accumulo/master/tableOps/MasterRepo.java | 49 +
.../accumulo/master/tableOps/RenameTable.java | 91 +
.../accumulo/master/tableOps/TableRangeOp.java | 136 ++
.../accumulo/master/tableOps/TraceRepo.java | 83 +
.../apache/accumulo/master/tableOps/Utils.java | 132 ++
.../master/tserverOps/ShutdownTServer.java | 94 +
.../apache/accumulo/master/util/FateAdmin.java | 88 +
.../apache/accumulo/master/DefaultMapTest.java | 36 +
.../apache/accumulo/master/TestMergeState.java | 195 ++
.../balancer/ChaoticLoadBalancerTest.java | 167 ++
.../balancer/DefaultLoadBalancerTest.java | 281 +++
.../master/balancer/TableLoadBalancerTest.java | 164 ++
.../accumulo/master/state/MergeInfoTest.java | 76 +
.../master/state/RootTabletStateStoreTest.java | 218 +++
.../org/apache/accumulo/monitor/Monitor.java | 62 +-
.../accumulo/monitor/servlets/JSONServlet.java | 3 +-
.../monitor/servlets/OperationServlet.java | 2 +-
.../monitor/servlets/TServersServlet.java | 5 +-
.../monitor/servlets/TablesServlet.java | 7 +-
.../accumulo/monitor/servlets/XMLServlet.java | 5 +-
server/pom.xml | 2 +
server/server/pom.xml | 8 +
.../server/data/ServerColumnUpdate.java | 36 -
.../accumulo/server/data/ServerMutation.java | 81 -
.../org/apache/accumulo/server/fate/Admin.java | 88 -
.../server/master/EventCoordinator.java | 63 -
.../accumulo/server/master/LiveTServerSet.java | 398 ----
.../apache/accumulo/server/master/Master.java | 1771 ------------------
.../server/master/TServerLockWatcher.java | 43 -
.../server/master/TabletGroupWatcher.java | 652 -------
.../master/balancer/ChaoticLoadBalancer.java | 143 --
.../master/balancer/DefaultLoadBalancer.java | 318 ----
.../master/balancer/TableLoadBalancer.java | 147 --
.../server/master/balancer/TabletBalancer.java | 151 --
.../server/master/recovery/HadoopLogCloser.java | 61 -
.../server/master/recovery/LogCloser.java | 27 -
.../server/master/recovery/MapRLogCloser.java | 47 -
.../server/master/recovery/RecoveryManager.java | 177 --
.../server/master/state/Assignment.java | 29 -
.../server/master/state/DistributedStore.java | 34 -
.../master/state/DistributedStoreException.java | 34 -
.../server/master/state/MergeStats.java | 254 ---
.../server/master/state/MetaDataStateStore.java | 157 --
.../master/state/RootTabletStateStore.java | 45 -
.../server/master/state/SetGoalState.java | 48 -
.../server/master/state/TableCounts.java | 37 -
.../server/master/state/TableStats.java | 70 -
.../server/master/state/TabletMigration.java | 35 -
.../server/master/state/TabletStateStore.java | 92 -
.../accumulo/server/master/state/ZooStore.java | 96 -
.../master/state/ZooTabletStateStore.java | 174 --
.../server/master/tableOps/BulkImport.java | 621 ------
.../master/tableOps/CancelCompactions.java | 99 -
.../master/tableOps/ChangeTableState.java | 64 -
.../server/master/tableOps/CloneTable.java | 243 ---
.../server/master/tableOps/CompactRange.java | 386 ----
.../server/master/tableOps/CreateTable.java | 288 ---
.../server/master/tableOps/DeleteTable.java | 242 ---
.../server/master/tableOps/ExportTable.java | 309 ---
.../server/master/tableOps/ImportTable.java | 601 ------
.../server/master/tableOps/MasterRepo.java | 49 -
.../server/master/tableOps/RenameTable.java | 91 -
.../server/master/tableOps/TableRangeOp.java | 136 --
.../server/master/tableOps/TraceRepo.java | 83 -
.../accumulo/server/master/tableOps/Utils.java | 132 --
.../master/tserverOps/ShutdownTServer.java | 94 -
.../accumulo/server/tabletserver/Tablet.java | 4 +-
.../server/tabletserver/TabletServer.java | 14 +-
.../server/tabletserver/TabletTime.java | 224 ---
.../tabletserver/UniqueNameAllocator.java | 79 -
.../server/tabletserver/log/DfsLogger.java | 2 +-
.../server/util/FindOfflineTablets.java | 16 +-
.../server/util/MasterMetadataUtil.java | 2 +-
.../accumulo/server/util/TabletOperations.java | 78 -
.../accumulo/server/master/DefaultMapTest.java | 36 -
.../accumulo/server/master/TestMergeState.java | 195 --
.../balancer/ChaoticLoadBalancerTest.java | 166 --
.../balancer/DefaultLoadBalancerTest.java | 281 ---
.../master/balancer/TableLoadBalancerTest.java | 164 --
.../server/master/state/MergeInfoTest.java | 76 -
.../master/state/RootTabletStateStoreTest.java | 212 ---
.../tabletserver/CheckTabletMetadataTest.java | 2 +-
server/tserver/pom.xml | 130 ++
.../apache/accumulo/test/GetMasterStats.java | 4 +-
.../continuous/ContinuousStatsCollector.java | 6 +-
.../test/functional/SplitRecoveryTest.java | 4 +-
.../metadata/MetadataBatchScanTest.java | 2 +-
.../test/performance/thrift/NullTserver.java | 10 +-
.../test/randomwalk/concurrent/Shutdown.java | 2 +-
.../test/randomwalk/concurrent/StartAll.java | 2 +-
.../randomwalk/concurrent/StopTabletServer.java | 2 +-
.../test/functional/ChaoticBalancerIT.java | 2 +-
.../test/functional/MasterFailoverIT.java | 2 +-
.../accumulo/test/functional/RestartIT.java | 22 +-
158 files changed, 11687 insertions(+), 11349 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java
----------------------------------------------------------------------
diff --git a/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java b/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java
index fc6c9c0..aff4271 100644
--- a/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java
+++ b/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java
@@ -47,8 +47,8 @@ import org.apache.accumulo.core.master.thrift.MasterGoalState;
import org.apache.accumulo.core.util.Daemon;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.core.util.UtilWaitThread;
-import org.apache.accumulo.server.master.Master;
-import org.apache.accumulo.server.master.state.SetGoalState;
+import org.apache.accumulo.master.Master;
+import org.apache.accumulo.master.state.SetGoalState;
import org.apache.accumulo.server.tabletserver.TabletServer;
import org.apache.accumulo.server.util.Initialize;
import org.apache.accumulo.server.util.PortUtils;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 4f81e51..4c1d9bf 100644
--- a/pom.xml
+++ b/pom.xml
@@ -236,6 +236,11 @@
</dependency>
<dependency>
<groupId>org.apache.accumulo</groupId>
+ <artifactId>accumulo-master</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-maven-plugin</artifactId>
<version>${project.version}</version>
</dependency>
@@ -280,6 +285,11 @@
<version>${project.version}</version>
</dependency>
<dependency>
+ <groupId>org.apache.accumulo</groupId>
+ <artifactId>accumulo-tserver</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-jci-core</artifactId>
<version>1.0</version>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/master/state/CurrentState.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/master/state/CurrentState.java b/server/base/src/main/java/org/apache/accumulo/master/state/CurrentState.java
new file mode 100644
index 0000000..8995fdb
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/master/state/CurrentState.java
@@ -0,0 +1,30 @@
+/*
+ * 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.accumulo.master.state;
+
+import java.util.Collection;
+import java.util.Set;
+
+public interface CurrentState {
+
+ Set<String> onlineTables();
+
+ Set<TServerInstance> onlineTabletServers();
+
+ Collection<MergeInfo> merges();
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/master/state/DeadServerList.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/master/state/DeadServerList.java b/server/base/src/main/java/org/apache/accumulo/master/state/DeadServerList.java
new file mode 100644
index 0000000..261f200
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/master/state/DeadServerList.java
@@ -0,0 +1,80 @@
+/*
+ * 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.accumulo.master.state;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.accumulo.core.master.thrift.DeadServer;
+import org.apache.accumulo.fate.zookeeper.IZooReaderWriter;
+import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeExistsPolicy;
+import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeMissingPolicy;
+import org.apache.accumulo.server.zookeeper.ZooReaderWriter;
+import org.apache.log4j.Logger;
+import org.apache.zookeeper.data.Stat;
+
+public class DeadServerList {
+ private static final Logger log = Logger.getLogger(DeadServerList.class);
+ private final String path;
+
+ public DeadServerList(String path) {
+ this.path = path;
+ IZooReaderWriter zoo = ZooReaderWriter.getInstance();
+ try {
+ zoo.mkdirs(path);
+ } catch (Exception ex) {
+ log.error("Unable to make parent directories of " + path, ex);
+ }
+ }
+
+ public List<DeadServer> getList() {
+ List<DeadServer> result = new ArrayList<DeadServer>();
+ IZooReaderWriter zoo = ZooReaderWriter.getInstance();
+ try {
+ List<String> children = zoo.getChildren(path);
+ if (children != null) {
+ for (String child : children) {
+ Stat stat = new Stat();
+ byte[] data = zoo.getData(path + "/" + child, stat);
+ DeadServer server = new DeadServer(child, stat.getMtime(), new String(data));
+ result.add(server);
+ }
+ }
+ } catch (Exception ex) {
+ log.error(ex, ex);
+ }
+ return result;
+ }
+
+ public void delete(String server) {
+ IZooReaderWriter zoo = ZooReaderWriter.getInstance();
+ try {
+ zoo.recursiveDelete(path + "/" + server, NodeMissingPolicy.SKIP);
+ } catch (Exception ex) {
+ log.error(ex, ex);
+ }
+ }
+
+ public void post(String server, String cause) {
+ IZooReaderWriter zoo = ZooReaderWriter.getInstance();
+ try {
+ zoo.putPersistentData(path + "/" + server, cause.getBytes(), NodeExistsPolicy.SKIP);
+ } catch (Exception ex) {
+ log.error(ex, ex);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/master/state/MergeInfo.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/master/state/MergeInfo.java b/server/base/src/main/java/org/apache/accumulo/master/state/MergeInfo.java
new file mode 100644
index 0000000..056836e
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/master/state/MergeInfo.java
@@ -0,0 +1,107 @@
+/*
+ * 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.accumulo.master.state;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.accumulo.core.data.KeyExtent;
+import org.apache.hadoop.io.Writable;
+
+/**
+ * Information about the current merge/rangeDelete.
+ *
+ * Writable to serialize for zookeeper and the Tablet
+ */
+public class MergeInfo implements Writable {
+
+ public enum Operation {
+ MERGE, DELETE,
+ }
+
+ MergeState state = MergeState.NONE;
+ KeyExtent extent;
+ Operation operation = Operation.MERGE;
+
+ public MergeInfo() {}
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ extent = new KeyExtent();
+ extent.readFields(in);
+ state = MergeState.values()[in.readInt()];
+ operation = Operation.values()[in.readInt()];
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ extent.write(out);
+ out.writeInt(state.ordinal());
+ out.writeInt(operation.ordinal());
+ }
+
+ public MergeInfo(KeyExtent range, Operation op) {
+ this.extent = range;
+ this.operation = op;
+ }
+
+ public MergeState getState() {
+ return state;
+ }
+
+ public KeyExtent getExtent() {
+ return extent;
+ }
+
+ public Operation getOperation() {
+ return operation;
+ }
+
+ public void setState(MergeState state) {
+ this.state = state;
+ }
+
+ public boolean isDelete() {
+ return this.operation.equals(Operation.DELETE);
+ }
+
+ public boolean needsToBeChopped(KeyExtent otherExtent) {
+ // During a delete, the block after the merge will be stretched to cover the deleted area.
+ // Therefore, it needs to be chopped
+ if (!otherExtent.getTableId().equals(extent.getTableId()))
+ return false;
+ if (isDelete())
+ return otherExtent.getPrevEndRow() != null && otherExtent.getPrevEndRow().equals(extent.getEndRow());
+ else
+ return this.extent.overlaps(otherExtent);
+ }
+
+ public boolean overlaps(KeyExtent otherExtent) {
+ boolean result = this.extent.overlaps(otherExtent);
+ if (!result && needsToBeChopped(otherExtent))
+ return true;
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ if (!state.equals(MergeState.NONE))
+ return "Merge " + operation.toString() + " of " + extent + " State: " + state;
+ return "No Merge in progress";
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/master/state/MergeState.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/master/state/MergeState.java b/server/base/src/main/java/org/apache/accumulo/master/state/MergeState.java
new file mode 100644
index 0000000..eb53c3a
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/master/state/MergeState.java
@@ -0,0 +1,49 @@
+/*
+ * 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.accumulo.master.state;
+
+public enum MergeState {
+ /**
+ * Not merging
+ */
+ NONE,
+ /**
+ * created, stored in zookeeper, other merges are prevented on the table
+ */
+ STARTED,
+ /**
+ * put all matching tablets online, split tablets if we are deleting
+ */
+ SPLITTING,
+ /**
+ * after the tablet server chops the file, it marks the metadata table with a chopped marker
+ */
+ WAITING_FOR_CHOPPED,
+ /**
+ * when the number of chopped tablets in the range matches the number of online tablets in the range, take the tablets offline
+ */
+ WAITING_FOR_OFFLINE,
+ /**
+ * when the number of chopped, offline tablets equals the number of merge tablets, begin the metadata updates
+ */
+ MERGING,
+ /**
+ * merge is complete, the resulting tablet can be brought online, remove the marker in zookeeper
+ */
+ COMPLETE;
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/master/state/MetaDataTableScanner.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/master/state/MetaDataTableScanner.java b/server/base/src/main/java/org/apache/accumulo/master/state/MetaDataTableScanner.java
new file mode 100644
index 0000000..a181325
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/master/state/MetaDataTableScanner.java
@@ -0,0 +1,196 @@
+/*
+ * 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.accumulo.master.state;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.SortedMap;
+
+import org.apache.accumulo.core.client.BatchScanner;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Instance;
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.ScannerBase;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.KeyExtent;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.user.WholeRowIterator;
+import org.apache.accumulo.core.metadata.MetadataTable;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ChoppedColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LogColumnFamily;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.security.Credentials;
+import org.apache.accumulo.master.state.TabletLocationState.BadLocationStateException;
+import org.apache.hadoop.io.Text;
+import org.apache.log4j.Logger;
+
+public class MetaDataTableScanner implements Iterator<TabletLocationState> {
+ private static final Logger log = Logger.getLogger(MetaDataTableScanner.class);
+
+ BatchScanner mdScanner;
+ Iterator<Entry<Key,Value>> iter;
+
+ public MetaDataTableScanner(Instance instance, Credentials credentials, Range range, CurrentState state) {
+ this(instance, credentials, range, state, MetadataTable.NAME);
+ }
+
+ MetaDataTableScanner(Instance instance, Credentials credentials, Range range, CurrentState state, String tableName) {
+ // scan over metadata table, looking for tablets in the wrong state based on the live servers and online tables
+ try {
+ Connector connector = instance.getConnector(credentials.getPrincipal(), credentials.getToken());
+ mdScanner = connector.createBatchScanner(tableName, Authorizations.EMPTY, 8);
+ configureScanner(mdScanner, state);
+ mdScanner.setRanges(Collections.singletonList(range));
+ iter = mdScanner.iterator();
+ } catch (Exception ex) {
+ mdScanner.close();
+ throw new RuntimeException(ex);
+ }
+ }
+
+ static public void configureScanner(ScannerBase scanner, CurrentState state) {
+ TabletsSection.TabletColumnFamily.PREV_ROW_COLUMN.fetch(scanner);
+ scanner.fetchColumnFamily(TabletsSection.CurrentLocationColumnFamily.NAME);
+ scanner.fetchColumnFamily(TabletsSection.FutureLocationColumnFamily.NAME);
+ scanner.fetchColumnFamily(LogColumnFamily.NAME);
+ scanner.fetchColumnFamily(ChoppedColumnFamily.NAME);
+ scanner.addScanIterator(new IteratorSetting(1000, "wholeRows", WholeRowIterator.class));
+ IteratorSetting tabletChange = new IteratorSetting(1001, "tabletChange", TabletStateChangeIterator.class);
+ if (state != null) {
+ TabletStateChangeIterator.setCurrentServers(tabletChange, state.onlineTabletServers());
+ TabletStateChangeIterator.setOnlineTables(tabletChange, state.onlineTables());
+ TabletStateChangeIterator.setMerges(tabletChange, state.merges());
+ }
+ scanner.addScanIterator(tabletChange);
+ }
+
+ public MetaDataTableScanner(Instance instance, Credentials credentials, Range range) {
+ this(instance, credentials, range, MetadataTable.NAME);
+ }
+
+ public MetaDataTableScanner(Instance instance, Credentials credentials, Range range, String tableName) {
+ this(instance, credentials, range, null, tableName);
+ }
+
+ public void close() {
+ if (iter != null) {
+ mdScanner.close();
+ iter = null;
+ }
+ }
+
+ @Override
+ public void finalize() {
+ close();
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (iter == null)
+ return false;
+ boolean result = iter.hasNext();
+ if (!result) {
+ close();
+ }
+ return result;
+ }
+
+ @Override
+ public TabletLocationState next() {
+ try {
+ return fetch();
+ } catch (RuntimeException ex) {
+ // something is wrong with the records in the !METADATA table, just skip over it
+ log.error(ex, ex);
+ mdScanner.close();
+ return null;
+ }
+ }
+
+ public static TabletLocationState createTabletLocationState(Key k, Value v) throws IOException, BadLocationStateException {
+ final SortedMap<Key,Value> decodedRow = WholeRowIterator.decodeRow(k, v);
+ KeyExtent extent = null;
+ TServerInstance future = null;
+ TServerInstance current = null;
+ TServerInstance last = null;
+ List<Collection<String>> walogs = new ArrayList<Collection<String>>();
+ boolean chopped = false;
+
+ for (Entry<Key,Value> entry : decodedRow.entrySet()) {
+ Key key = entry.getKey();
+ Text row = key.getRow();
+ Text cf = key.getColumnFamily();
+ Text cq = key.getColumnQualifier();
+
+ if (cf.compareTo(TabletsSection.FutureLocationColumnFamily.NAME) == 0) {
+ TServerInstance location = new TServerInstance(entry.getValue(), cq);
+ if (future != null) {
+ throw new BadLocationStateException("found two assignments for the same extent " + key.getRow() + ": " + future + " and " + location);
+ }
+ future = location;
+ } else if (cf.compareTo(TabletsSection.CurrentLocationColumnFamily.NAME) == 0) {
+ TServerInstance location = new TServerInstance(entry.getValue(), cq);
+ if (current != null) {
+ throw new BadLocationStateException("found two locations for the same extent " + key.getRow() + ": " + current + " and " + location);
+ }
+ current = location;
+ } else if (cf.compareTo(LogColumnFamily.NAME) == 0) {
+ String[] split = entry.getValue().toString().split("\\|")[0].split(";");
+ walogs.add(Arrays.asList(split));
+ } else if (cf.compareTo(TabletsSection.LastLocationColumnFamily.NAME) == 0) {
+ TServerInstance location = new TServerInstance(entry.getValue(), cq);
+ if (last != null) {
+ throw new BadLocationStateException("found two last locations for the same extent " + key.getRow() + ": " + last + " and " + location);
+ }
+ last = new TServerInstance(entry.getValue(), cq);
+ } else if (cf.compareTo(ChoppedColumnFamily.NAME) == 0) {
+ chopped = true;
+ } else if (TabletsSection.TabletColumnFamily.PREV_ROW_COLUMN.equals(cf, cq)) {
+ extent = new KeyExtent(row, entry.getValue());
+ }
+ }
+ if (extent == null) {
+ log.warn("No prev-row for key extent: " + decodedRow);
+ return null;
+ }
+ return new TabletLocationState(extent, future, current, last, walogs, chopped);
+ }
+
+ private TabletLocationState fetch() {
+ try {
+ Entry<Key,Value> e = iter.next();
+ return createTabletLocationState(e.getKey(), e.getValue());
+ } catch (IOException ex) {
+ throw new RuntimeException(ex);
+ } catch (BadLocationStateException ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+
+ @Override
+ public void remove() {
+ throw new RuntimeException("Unimplemented");
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/master/state/TServerInstance.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/master/state/TServerInstance.java b/server/base/src/main/java/org/apache/accumulo/master/state/TServerInstance.java
new file mode 100644
index 0000000..f9c054a
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/master/state/TServerInstance.java
@@ -0,0 +1,127 @@
+/*
+ * 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.accumulo.master.state;
+
+import java.io.Serializable;
+import java.net.InetSocketAddress;
+
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
+import org.apache.accumulo.core.util.AddressUtil;
+import org.apache.hadoop.io.Text;
+
+/**
+ * A tablet is assigned to a tablet server at the given address as long as it is alive and well. When the tablet server is restarted, the instance information
+ * it advertises will change. Therefore tablet assignments can be considered out-of-date if the tablet server instance information has been changed.
+ *
+ */
+public class TServerInstance implements Comparable<TServerInstance>, Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private InetSocketAddress location;
+ private String session;
+ private String cachedStringRepresentation;
+
+ public TServerInstance(InetSocketAddress address, String session) {
+ this.location = address;
+ this.session = session;
+ this.cachedStringRepresentation = hostPort() + "[" + session + "]";
+ }
+
+ public TServerInstance(InetSocketAddress address, long session) {
+ this(address, Long.toHexString(session));
+ }
+
+ public TServerInstance(String address, long session) {
+ this(AddressUtil.parseAddress(address), Long.toHexString(session));
+ }
+
+ public TServerInstance(Value address, Text session) {
+ this(AddressUtil.parseAddress(new String(address.get())), session.toString());
+ }
+
+ public void putLocation(Mutation m) {
+ m.put(TabletsSection.CurrentLocationColumnFamily.NAME, asColumnQualifier(), asMutationValue());
+ }
+
+ public void putFutureLocation(Mutation m) {
+ m.put(TabletsSection.FutureLocationColumnFamily.NAME, asColumnQualifier(), asMutationValue());
+ }
+
+ public void putLastLocation(Mutation m) {
+ m.put(TabletsSection.LastLocationColumnFamily.NAME, asColumnQualifier(), asMutationValue());
+ }
+
+ public void clearLastLocation(Mutation m) {
+ m.putDelete(TabletsSection.LastLocationColumnFamily.NAME, asColumnQualifier());
+ }
+
+ @Override
+ public int compareTo(TServerInstance other) {
+ if (this == other)
+ return 0;
+ return this.toString().compareTo(other.toString());
+ }
+
+ @Override
+ public int hashCode() {
+ return toString().hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof TServerInstance) {
+ return compareTo((TServerInstance) obj) == 0;
+ }
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ return cachedStringRepresentation;
+ }
+
+ public int port() {
+ return getLocation().getPort();
+ }
+
+ public String host() {
+ return getLocation().getAddress().getHostAddress();
+ }
+
+ public String hostPort() {
+ return org.apache.accumulo.core.util.AddressUtil.toString(getLocation());
+ }
+
+ public Text asColumnQualifier() {
+ return new Text(this.getSession());
+ }
+
+ public Value asMutationValue() {
+ return new Value(org.apache.accumulo.core.util.AddressUtil.toString(getLocation()).getBytes());
+ }
+
+ public InetSocketAddress getLocation() {
+ return location;
+ }
+
+ public String getSession() {
+ return session;
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/master/state/TabletLocationState.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/master/state/TabletLocationState.java b/server/base/src/main/java/org/apache/accumulo/master/state/TabletLocationState.java
new file mode 100644
index 0000000..0a95c49
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/master/state/TabletLocationState.java
@@ -0,0 +1,97 @@
+/*
+ * 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.accumulo.master.state;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Set;
+
+import org.apache.accumulo.core.data.KeyExtent;
+
+/**
+ * When a tablet is assigned, we mark its future location. When the tablet is opened, we set its current location. A tablet should never have both a future and
+ * current location.
+ *
+ * A tablet server is always associated with a unique session id. If the current tablet server has a different session, we know the location information is
+ * out-of-date.
+ */
+public class TabletLocationState {
+
+ static public class BadLocationStateException extends Exception {
+ private static final long serialVersionUID = 1L;
+
+ BadLocationStateException(String msg) { super(msg); }
+ }
+
+ public TabletLocationState(KeyExtent extent, TServerInstance future, TServerInstance current, TServerInstance last, Collection<Collection<String>> walogs,
+ boolean chopped) throws BadLocationStateException {
+ this.extent = extent;
+ this.future = future;
+ this.current = current;
+ this.last = last;
+ if (walogs == null)
+ walogs = Collections.emptyList();
+ this.walogs = walogs;
+ this.chopped = chopped;
+ if (current != null && future != null) {
+ throw new BadLocationStateException(extent + " is both assigned and hosted, which should never happen: " + this);
+ }
+ }
+
+ final public KeyExtent extent;
+ final public TServerInstance future;
+ final public TServerInstance current;
+ final public TServerInstance last;
+ final public Collection<Collection<String>> walogs;
+ final public boolean chopped;
+
+ public String toString() {
+ return extent + "@(" + future + "," + current + "," + last + ")" + (chopped ? " chopped" : "");
+ }
+
+ public TServerInstance getServer() {
+ TServerInstance result = null;
+ if (current != null) {
+ result = current;
+ } else if (future != null) {
+ result = future;
+ } else {
+ result = last;
+ }
+ return result;
+ }
+
+ public TabletState getState(Set<TServerInstance> liveServers) {
+ TServerInstance server = getServer();
+ if (server == null)
+ return TabletState.UNASSIGNED;
+ if (server.equals(current) || server.equals(future)) {
+ if (liveServers.contains(server))
+ if (server.equals(future)) {
+ return TabletState.ASSIGNED;
+ } else {
+ return TabletState.HOSTED;
+ }
+ else {
+ return TabletState.ASSIGNED_TO_DEAD_SERVER;
+ }
+ }
+ // server == last
+ return TabletState.UNASSIGNED;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/master/state/TabletServerState.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/master/state/TabletServerState.java b/server/base/src/main/java/org/apache/accumulo/master/state/TabletServerState.java
new file mode 100644
index 0000000..41a54f3
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/master/state/TabletServerState.java
@@ -0,0 +1,76 @@
+/*
+ * 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.accumulo.master.state;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Set;
+
+public enum TabletServerState {
+ // not a valid state, reserved for internal use only
+ RESERVED((byte) (-1)),
+
+ // the following are normally functioning states
+ NEW((byte) 0),
+ ONLINE((byte) 1),
+ UNRESPONSIVE((byte) 2),
+ DOWN((byte) 3),
+
+ // the following are bad states and cause tservers to be ignored by the master
+ BAD_SYSTEM_PASSWORD((byte) 101),
+ BAD_VERSION((byte) 102),
+ BAD_INSTANCE((byte) 103),
+ BAD_CONFIG((byte) 104),
+ BAD_VERSION_AND_INSTANCE((byte) 105),
+ BAD_VERSION_AND_CONFIG((byte) 106),
+ BAD_VERSION_AND_INSTANCE_AND_CONFIG((byte) 107),
+ BAD_INSTANCE_AND_CONFIG((byte) 108);
+
+ private byte id;
+
+ private static HashMap<Byte,TabletServerState> mapping;
+ private static HashSet<TabletServerState> badStates;
+
+ static {
+ mapping = new HashMap<Byte,TabletServerState>(TabletServerState.values().length);
+ badStates = new HashSet<TabletServerState>();
+ for (TabletServerState state : TabletServerState.values()) {
+ mapping.put(state.id, state);
+ if (state.id > 99)
+ badStates.add(state);
+ }
+ }
+
+ private TabletServerState(byte id) {
+ this.id = id;
+ }
+
+ public byte getId() {
+ return this.id;
+ }
+
+ public static TabletServerState getStateById(byte id) {
+ if (mapping.containsKey(id))
+ return mapping.get(id);
+ throw new IndexOutOfBoundsException("No such state");
+ }
+
+ public static Set<TabletServerState> getBadStates() {
+ return Collections.unmodifiableSet(badStates);
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/master/state/TabletState.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/master/state/TabletState.java b/server/base/src/main/java/org/apache/accumulo/master/state/TabletState.java
new file mode 100644
index 0000000..2057a0e
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/master/state/TabletState.java
@@ -0,0 +1,21 @@
+/*
+ * 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.accumulo.master.state;
+
+public enum TabletState {
+ UNASSIGNED, ASSIGNED, HOSTED, ASSIGNED_TO_DEAD_SERVER
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/master/state/TabletStateChangeIterator.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/master/state/TabletStateChangeIterator.java b/server/base/src/main/java/org/apache/accumulo/master/state/TabletStateChangeIterator.java
new file mode 100644
index 0000000..cd87c1e
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/master/state/TabletStateChangeIterator.java
@@ -0,0 +1,188 @@
+/*
+ * 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.accumulo.master.state;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.KeyExtent;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.IteratorEnvironment;
+import org.apache.accumulo.core.iterators.SkippingIterator;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.apache.accumulo.core.util.AddressUtil;
+import org.apache.accumulo.core.util.StringUtil;
+import org.apache.accumulo.master.state.TabletLocationState.BadLocationStateException;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.io.Text;
+
+public class TabletStateChangeIterator extends SkippingIterator {
+
+ private static final String SERVERS_OPTION = "servers";
+ private static final String TABLES_OPTION = "tables";
+ private static final String MERGES_OPTION = "merges";
+ // private static final Logger log = Logger.getLogger(TabletStateChangeIterator.class);
+
+ Set<TServerInstance> current;
+ Set<String> onlineTables;
+ Map<Text,MergeInfo> merges;
+
+ @Override
+ public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options, IteratorEnvironment env) throws IOException {
+ super.init(source, options, env);
+ current = parseServers(options.get(SERVERS_OPTION));
+ onlineTables = parseTables(options.get(TABLES_OPTION));
+ merges = parseMerges(options.get(MERGES_OPTION));
+ }
+
+ private Set<String> parseTables(String tables) {
+ if (tables == null)
+ return null;
+ Set<String> result = new HashSet<String>();
+ for (String table : tables.split(","))
+ result.add(table);
+ return result;
+ }
+
+ private Set<TServerInstance> parseServers(String servers) {
+ if (servers == null)
+ return null;
+ // parse "host:port[INSTANCE]"
+ Set<TServerInstance> result = new HashSet<TServerInstance>();
+ if (servers.length() > 0) {
+ for (String part : servers.split(",")) {
+ String parts[] = part.split("\\[", 2);
+ String hostport = parts[0];
+ String instance = parts[1];
+ if (instance != null && instance.endsWith("]"))
+ instance = instance.substring(0, instance.length() - 1);
+ result.add(new TServerInstance(AddressUtil.parseAddress(hostport), instance));
+ }
+ }
+ return result;
+ }
+
+ private Map<Text,MergeInfo> parseMerges(String merges) {
+ if (merges == null)
+ return null;
+ try {
+ Map<Text,MergeInfo> result = new HashMap<Text,MergeInfo>();
+ DataInputBuffer buffer = new DataInputBuffer();
+ byte[] data = Base64.decodeBase64(merges.getBytes());
+ buffer.reset(data, data.length);
+ while (buffer.available() > 0) {
+ MergeInfo mergeInfo = new MergeInfo();
+ mergeInfo.readFields(buffer);
+ result.put(mergeInfo.extent.getTableId(), mergeInfo);
+ }
+ return result;
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+
+ @Override
+ protected void consume() throws IOException {
+ while (getSource().hasTop()) {
+ Key k = getSource().getTopKey();
+ Value v = getSource().getTopValue();
+
+ if (onlineTables == null || current == null)
+ return;
+
+ TabletLocationState tls;
+ try {
+ tls = MetaDataTableScanner.createTabletLocationState(k, v);
+ if (tls == null)
+ return;
+ } catch (BadLocationStateException e) {
+ // maybe the master can do something with a tablet with bad/inconsistent state
+ return;
+ }
+ // we always want data about merges
+ MergeInfo merge = merges.get(tls.extent.getTableId());
+ if (merge != null && merge.getExtent() != null && merge.getExtent().overlaps(tls.extent)) {
+ return;
+ }
+ // is the table supposed to be online or offline?
+ boolean shouldBeOnline = onlineTables.contains(tls.extent.getTableId().toString());
+
+ switch (tls.getState(current)) {
+ case ASSIGNED:
+ // we always want data about assigned tablets
+ return;
+ case HOSTED:
+ if (!shouldBeOnline)
+ return;
+ case ASSIGNED_TO_DEAD_SERVER:
+ return;
+ case UNASSIGNED:
+ if (shouldBeOnline)
+ return;
+ }
+ // table is in the expected state so don't bother returning any information about it
+ getSource().next();
+ }
+ }
+
+ @Override
+ public SortedKeyValueIterator<Key,Value> deepCopy(IteratorEnvironment env) {
+ throw new UnsupportedOperationException();
+ }
+
+ public static void setCurrentServers(IteratorSetting cfg, Set<TServerInstance> goodServers) {
+ if (goodServers != null) {
+ List<String> servers = new ArrayList<String>();
+ for (TServerInstance server : goodServers)
+ servers.add(server.toString());
+ cfg.addOption(SERVERS_OPTION, StringUtil.join(servers, ","));
+ }
+ }
+
+ public static void setOnlineTables(IteratorSetting cfg, Set<String> onlineTables) {
+ if (onlineTables != null)
+ cfg.addOption(TABLES_OPTION, StringUtil.join(onlineTables, ","));
+ }
+
+ public static void setMerges(IteratorSetting cfg, Collection<MergeInfo> merges) {
+ DataOutputBuffer buffer = new DataOutputBuffer();
+ try {
+ for (MergeInfo info : merges) {
+ KeyExtent extent = info.getExtent();
+ if (extent != null && !info.getState().equals(MergeState.NONE)) {
+ info.write(buffer);
+ }
+ }
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ String encoded = new String(Base64.encodeBase64(Arrays.copyOf(buffer.getData(), buffer.getLength())));
+ cfg.addOption(MERGES_OPTION, encoded);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/server/data/ServerColumnUpdate.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/data/ServerColumnUpdate.java b/server/base/src/main/java/org/apache/accumulo/server/data/ServerColumnUpdate.java
new file mode 100644
index 0000000..af992a6
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/server/data/ServerColumnUpdate.java
@@ -0,0 +1,36 @@
+/*
+ * 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.accumulo.server.data;
+
+import org.apache.accumulo.core.data.ColumnUpdate;
+
+public class ServerColumnUpdate extends ColumnUpdate {
+
+ ServerMutation parent;
+
+ public ServerColumnUpdate(byte[] cf, byte[] cq, byte[] cv, boolean hasts, long ts, boolean deleted, byte[] val, ServerMutation serverMutation) {
+ super(cf, cq, cv, hasts, ts, deleted, val);
+ parent = serverMutation;
+ }
+
+ public long getTimestamp() {
+ if (hasTimestamp())
+ return super.getTimestamp();
+ return parent.getSystemTimestamp();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/server/data/ServerMutation.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/data/ServerMutation.java b/server/base/src/main/java/org/apache/accumulo/server/data/ServerMutation.java
new file mode 100644
index 0000000..28a3515
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/server/data/ServerMutation.java
@@ -0,0 +1,81 @@
+/*
+ * 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.accumulo.server.data;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.accumulo.core.data.ColumnUpdate;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.thrift.TMutation;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.WritableUtils;
+
+/**
+ * Mutation that holds system time as computed by the tablet server when not provided by the user.
+ */
+public class ServerMutation extends Mutation {
+ private long systemTime = 0l;
+
+ public ServerMutation(TMutation tmutation) {
+ super(tmutation);
+ }
+
+ public ServerMutation(Text key) {
+ super(key);
+ }
+
+ public ServerMutation() {
+ }
+
+ protected void droppingOldTimestamp(long ts) {
+ this.systemTime = ts;
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ super.readFields(in);
+ // new format writes system time with the mutation
+ if (getSerializedFormat() == SERIALIZED_FORMAT.VERSION2)
+ systemTime = WritableUtils.readVLong(in);
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ super.write(out);
+ WritableUtils.writeVLong(out, systemTime);
+ }
+
+ public void setSystemTimestamp(long v) {
+ this.systemTime = v;
+ }
+
+ public long getSystemTimestamp() {
+ return this.systemTime;
+ }
+
+ @Override
+ protected ColumnUpdate newColumnUpdate(byte[] cf, byte[] cq, byte[] cv, boolean hasts, long ts, boolean deleted, byte[] val) {
+ return new ServerColumnUpdate(cf, cq, cv, hasts, ts, deleted, val, this);
+ }
+
+ @Override
+ public long estimatedMemoryUsed() {
+ return super.estimatedMemoryUsed() + 8;
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/server/master/state/CurrentState.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/master/state/CurrentState.java b/server/base/src/main/java/org/apache/accumulo/server/master/state/CurrentState.java
deleted file mode 100644
index f4d98bf..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/master/state/CurrentState.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * 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.accumulo.server.master.state;
-
-import java.util.Collection;
-import java.util.Set;
-
-public interface CurrentState {
-
- Set<String> onlineTables();
-
- Set<TServerInstance> onlineTabletServers();
-
- Collection<MergeInfo> merges();
-
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/server/master/state/DeadServerList.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/master/state/DeadServerList.java b/server/base/src/main/java/org/apache/accumulo/server/master/state/DeadServerList.java
deleted file mode 100644
index b2ea7d6..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/master/state/DeadServerList.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * 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.accumulo.server.master.state;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.accumulo.core.master.thrift.DeadServer;
-import org.apache.accumulo.fate.zookeeper.IZooReaderWriter;
-import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeExistsPolicy;
-import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeMissingPolicy;
-import org.apache.accumulo.server.zookeeper.ZooReaderWriter;
-import org.apache.log4j.Logger;
-import org.apache.zookeeper.data.Stat;
-
-public class DeadServerList {
- private static final Logger log = Logger.getLogger(DeadServerList.class);
- private final String path;
-
- public DeadServerList(String path) {
- this.path = path;
- IZooReaderWriter zoo = ZooReaderWriter.getInstance();
- try {
- zoo.mkdirs(path);
- } catch (Exception ex) {
- log.error("Unable to make parent directories of " + path, ex);
- }
- }
-
- public List<DeadServer> getList() {
- List<DeadServer> result = new ArrayList<DeadServer>();
- IZooReaderWriter zoo = ZooReaderWriter.getInstance();
- try {
- List<String> children = zoo.getChildren(path);
- if (children != null) {
- for (String child : children) {
- Stat stat = new Stat();
- byte[] data = zoo.getData(path + "/" + child, stat);
- DeadServer server = new DeadServer(child, stat.getMtime(), new String(data));
- result.add(server);
- }
- }
- } catch (Exception ex) {
- log.error(ex, ex);
- }
- return result;
- }
-
- public void delete(String server) {
- IZooReaderWriter zoo = ZooReaderWriter.getInstance();
- try {
- zoo.recursiveDelete(path + "/" + server, NodeMissingPolicy.SKIP);
- } catch (Exception ex) {
- log.error(ex, ex);
- }
- }
-
- public void post(String server, String cause) {
- IZooReaderWriter zoo = ZooReaderWriter.getInstance();
- try {
- zoo.putPersistentData(path + "/" + server, cause.getBytes(), NodeExistsPolicy.SKIP);
- } catch (Exception ex) {
- log.error(ex, ex);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/server/master/state/MergeInfo.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/master/state/MergeInfo.java b/server/base/src/main/java/org/apache/accumulo/server/master/state/MergeInfo.java
deleted file mode 100644
index 708b1b7..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/master/state/MergeInfo.java
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
- * 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.accumulo.server.master.state;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-
-import org.apache.accumulo.core.data.KeyExtent;
-import org.apache.hadoop.io.Writable;
-
-/**
- * Information about the current merge/rangeDelete.
- *
- * Writable to serialize for zookeeper and the Tablet
- */
-public class MergeInfo implements Writable {
-
- public enum Operation {
- MERGE, DELETE,
- }
-
- MergeState state = MergeState.NONE;
- KeyExtent extent;
- Operation operation = Operation.MERGE;
-
- public MergeInfo() {}
-
- @Override
- public void readFields(DataInput in) throws IOException {
- extent = new KeyExtent();
- extent.readFields(in);
- state = MergeState.values()[in.readInt()];
- operation = Operation.values()[in.readInt()];
- }
-
- @Override
- public void write(DataOutput out) throws IOException {
- extent.write(out);
- out.writeInt(state.ordinal());
- out.writeInt(operation.ordinal());
- }
-
- public MergeInfo(KeyExtent range, Operation op) {
- this.extent = range;
- this.operation = op;
- }
-
- public MergeState getState() {
- return state;
- }
-
- public KeyExtent getExtent() {
- return extent;
- }
-
- public Operation getOperation() {
- return operation;
- }
-
- public void setState(MergeState state) {
- this.state = state;
- }
-
- public boolean isDelete() {
- return this.operation.equals(Operation.DELETE);
- }
-
- public boolean needsToBeChopped(KeyExtent otherExtent) {
- // During a delete, the block after the merge will be stretched to cover the deleted area.
- // Therefore, it needs to be chopped
- if (!otherExtent.getTableId().equals(extent.getTableId()))
- return false;
- if (isDelete())
- return otherExtent.getPrevEndRow() != null && otherExtent.getPrevEndRow().equals(extent.getEndRow());
- else
- return this.extent.overlaps(otherExtent);
- }
-
- public boolean overlaps(KeyExtent otherExtent) {
- boolean result = this.extent.overlaps(otherExtent);
- if (!result && needsToBeChopped(otherExtent))
- return true;
- return result;
- }
-
- @Override
- public String toString() {
- if (!state.equals(MergeState.NONE))
- return "Merge " + operation.toString() + " of " + extent + " State: " + state;
- return "No Merge in progress";
- }
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/server/master/state/MergeState.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/master/state/MergeState.java b/server/base/src/main/java/org/apache/accumulo/server/master/state/MergeState.java
deleted file mode 100644
index 29b6ae3..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/master/state/MergeState.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * 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.accumulo.server.master.state;
-
-public enum MergeState {
- /**
- * Not merging
- */
- NONE,
- /**
- * created, stored in zookeeper, other merges are prevented on the table
- */
- STARTED,
- /**
- * put all matching tablets online, split tablets if we are deleting
- */
- SPLITTING,
- /**
- * after the tablet server chops the file, it marks the metadata table with a chopped marker
- */
- WAITING_FOR_CHOPPED,
- /**
- * when the number of chopped tablets in the range matches the number of online tablets in the range, take the tablets offline
- */
- WAITING_FOR_OFFLINE,
- /**
- * when the number of chopped, offline tablets equals the number of merge tablets, begin the metadata updates
- */
- MERGING,
- /**
- * merge is complete, the resulting tablet can be brought online, remove the marker in zookeeper
- */
- COMPLETE;
-
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/server/master/state/MetaDataTableScanner.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/master/state/MetaDataTableScanner.java b/server/base/src/main/java/org/apache/accumulo/server/master/state/MetaDataTableScanner.java
deleted file mode 100644
index cec0bcf..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/master/state/MetaDataTableScanner.java
+++ /dev/null
@@ -1,196 +0,0 @@
-/*
- * 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.accumulo.server.master.state;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map.Entry;
-import java.util.SortedMap;
-
-import org.apache.accumulo.core.client.BatchScanner;
-import org.apache.accumulo.core.client.Connector;
-import org.apache.accumulo.core.client.Instance;
-import org.apache.accumulo.core.client.IteratorSetting;
-import org.apache.accumulo.core.client.ScannerBase;
-import org.apache.accumulo.core.data.Key;
-import org.apache.accumulo.core.data.KeyExtent;
-import org.apache.accumulo.core.data.Range;
-import org.apache.accumulo.core.data.Value;
-import org.apache.accumulo.core.iterators.user.WholeRowIterator;
-import org.apache.accumulo.core.metadata.MetadataTable;
-import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
-import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ChoppedColumnFamily;
-import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LogColumnFamily;
-import org.apache.accumulo.core.security.Authorizations;
-import org.apache.accumulo.core.security.Credentials;
-import org.apache.accumulo.server.master.state.TabletLocationState.BadLocationStateException;
-import org.apache.hadoop.io.Text;
-import org.apache.log4j.Logger;
-
-public class MetaDataTableScanner implements Iterator<TabletLocationState> {
- private static final Logger log = Logger.getLogger(MetaDataTableScanner.class);
-
- BatchScanner mdScanner;
- Iterator<Entry<Key,Value>> iter;
-
- public MetaDataTableScanner(Instance instance, Credentials credentials, Range range, CurrentState state) {
- this(instance, credentials, range, state, MetadataTable.NAME);
- }
-
- MetaDataTableScanner(Instance instance, Credentials credentials, Range range, CurrentState state, String tableName) {
- // scan over metadata table, looking for tablets in the wrong state based on the live servers and online tables
- try {
- Connector connector = instance.getConnector(credentials.getPrincipal(), credentials.getToken());
- mdScanner = connector.createBatchScanner(tableName, Authorizations.EMPTY, 8);
- configureScanner(mdScanner, state);
- mdScanner.setRanges(Collections.singletonList(range));
- iter = mdScanner.iterator();
- } catch (Exception ex) {
- mdScanner.close();
- throw new RuntimeException(ex);
- }
- }
-
- static public void configureScanner(ScannerBase scanner, CurrentState state) {
- TabletsSection.TabletColumnFamily.PREV_ROW_COLUMN.fetch(scanner);
- scanner.fetchColumnFamily(TabletsSection.CurrentLocationColumnFamily.NAME);
- scanner.fetchColumnFamily(TabletsSection.FutureLocationColumnFamily.NAME);
- scanner.fetchColumnFamily(LogColumnFamily.NAME);
- scanner.fetchColumnFamily(ChoppedColumnFamily.NAME);
- scanner.addScanIterator(new IteratorSetting(1000, "wholeRows", WholeRowIterator.class));
- IteratorSetting tabletChange = new IteratorSetting(1001, "tabletChange", TabletStateChangeIterator.class);
- if (state != null) {
- TabletStateChangeIterator.setCurrentServers(tabletChange, state.onlineTabletServers());
- TabletStateChangeIterator.setOnlineTables(tabletChange, state.onlineTables());
- TabletStateChangeIterator.setMerges(tabletChange, state.merges());
- }
- scanner.addScanIterator(tabletChange);
- }
-
- public MetaDataTableScanner(Instance instance, Credentials credentials, Range range) {
- this(instance, credentials, range, MetadataTable.NAME);
- }
-
- public MetaDataTableScanner(Instance instance, Credentials credentials, Range range, String tableName) {
- this(instance, credentials, range, null, tableName);
- }
-
- public void close() {
- if (iter != null) {
- mdScanner.close();
- iter = null;
- }
- }
-
- @Override
- public void finalize() {
- close();
- }
-
- @Override
- public boolean hasNext() {
- if (iter == null)
- return false;
- boolean result = iter.hasNext();
- if (!result) {
- close();
- }
- return result;
- }
-
- @Override
- public TabletLocationState next() {
- try {
- return fetch();
- } catch (RuntimeException ex) {
- // something is wrong with the records in the !METADATA table, just skip over it
- log.error(ex, ex);
- mdScanner.close();
- return null;
- }
- }
-
- public static TabletLocationState createTabletLocationState(Key k, Value v) throws IOException, BadLocationStateException {
- final SortedMap<Key,Value> decodedRow = WholeRowIterator.decodeRow(k, v);
- KeyExtent extent = null;
- TServerInstance future = null;
- TServerInstance current = null;
- TServerInstance last = null;
- List<Collection<String>> walogs = new ArrayList<Collection<String>>();
- boolean chopped = false;
-
- for (Entry<Key,Value> entry : decodedRow.entrySet()) {
- Key key = entry.getKey();
- Text row = key.getRow();
- Text cf = key.getColumnFamily();
- Text cq = key.getColumnQualifier();
-
- if (cf.compareTo(TabletsSection.FutureLocationColumnFamily.NAME) == 0) {
- TServerInstance location = new TServerInstance(entry.getValue(), cq);
- if (future != null) {
- throw new BadLocationStateException("found two assignments for the same extent " + key.getRow() + ": " + future + " and " + location);
- }
- future = location;
- } else if (cf.compareTo(TabletsSection.CurrentLocationColumnFamily.NAME) == 0) {
- TServerInstance location = new TServerInstance(entry.getValue(), cq);
- if (current != null) {
- throw new BadLocationStateException("found two locations for the same extent " + key.getRow() + ": " + current + " and " + location);
- }
- current = location;
- } else if (cf.compareTo(LogColumnFamily.NAME) == 0) {
- String[] split = entry.getValue().toString().split("\\|")[0].split(";");
- walogs.add(Arrays.asList(split));
- } else if (cf.compareTo(TabletsSection.LastLocationColumnFamily.NAME) == 0) {
- TServerInstance location = new TServerInstance(entry.getValue(), cq);
- if (last != null) {
- throw new BadLocationStateException("found two last locations for the same extent " + key.getRow() + ": " + last + " and " + location);
- }
- last = new TServerInstance(entry.getValue(), cq);
- } else if (cf.compareTo(ChoppedColumnFamily.NAME) == 0) {
- chopped = true;
- } else if (TabletsSection.TabletColumnFamily.PREV_ROW_COLUMN.equals(cf, cq)) {
- extent = new KeyExtent(row, entry.getValue());
- }
- }
- if (extent == null) {
- log.warn("No prev-row for key extent: " + decodedRow);
- return null;
- }
- return new TabletLocationState(extent, future, current, last, walogs, chopped);
- }
-
- private TabletLocationState fetch() {
- try {
- Entry<Key,Value> e = iter.next();
- return createTabletLocationState(e.getKey(), e.getValue());
- } catch (IOException ex) {
- throw new RuntimeException(ex);
- } catch (BadLocationStateException ex) {
- throw new RuntimeException(ex);
- }
- }
-
- @Override
- public void remove() {
- throw new RuntimeException("Unimplemented");
- }
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/server/master/state/TServerInstance.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/master/state/TServerInstance.java b/server/base/src/main/java/org/apache/accumulo/server/master/state/TServerInstance.java
deleted file mode 100644
index 753a3cf..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/master/state/TServerInstance.java
+++ /dev/null
@@ -1,127 +0,0 @@
-/*
- * 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.accumulo.server.master.state;
-
-import java.io.Serializable;
-import java.net.InetSocketAddress;
-
-import org.apache.accumulo.core.data.Mutation;
-import org.apache.accumulo.core.data.Value;
-import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
-import org.apache.accumulo.core.util.AddressUtil;
-import org.apache.hadoop.io.Text;
-
-/**
- * A tablet is assigned to a tablet server at the given address as long as it is alive and well. When the tablet server is restarted, the instance information
- * it advertises will change. Therefore tablet assignments can be considered out-of-date if the tablet server instance information has been changed.
- *
- */
-public class TServerInstance implements Comparable<TServerInstance>, Serializable {
-
- private static final long serialVersionUID = 1L;
-
- private InetSocketAddress location;
- private String session;
- private String cachedStringRepresentation;
-
- public TServerInstance(InetSocketAddress address, String session) {
- this.location = address;
- this.session = session;
- this.cachedStringRepresentation = hostPort() + "[" + session + "]";
- }
-
- public TServerInstance(InetSocketAddress address, long session) {
- this(address, Long.toHexString(session));
- }
-
- public TServerInstance(String address, long session) {
- this(AddressUtil.parseAddress(address), Long.toHexString(session));
- }
-
- public TServerInstance(Value address, Text session) {
- this(AddressUtil.parseAddress(new String(address.get())), session.toString());
- }
-
- public void putLocation(Mutation m) {
- m.put(TabletsSection.CurrentLocationColumnFamily.NAME, asColumnQualifier(), asMutationValue());
- }
-
- public void putFutureLocation(Mutation m) {
- m.put(TabletsSection.FutureLocationColumnFamily.NAME, asColumnQualifier(), asMutationValue());
- }
-
- public void putLastLocation(Mutation m) {
- m.put(TabletsSection.LastLocationColumnFamily.NAME, asColumnQualifier(), asMutationValue());
- }
-
- public void clearLastLocation(Mutation m) {
- m.putDelete(TabletsSection.LastLocationColumnFamily.NAME, asColumnQualifier());
- }
-
- @Override
- public int compareTo(TServerInstance other) {
- if (this == other)
- return 0;
- return this.toString().compareTo(other.toString());
- }
-
- @Override
- public int hashCode() {
- return toString().hashCode();
- }
-
- @Override
- public boolean equals(Object obj) {
- if (obj instanceof TServerInstance) {
- return compareTo((TServerInstance) obj) == 0;
- }
- return false;
- }
-
- @Override
- public String toString() {
- return cachedStringRepresentation;
- }
-
- public int port() {
- return getLocation().getPort();
- }
-
- public String host() {
- return getLocation().getAddress().getHostAddress();
- }
-
- public String hostPort() {
- return org.apache.accumulo.core.util.AddressUtil.toString(getLocation());
- }
-
- public Text asColumnQualifier() {
- return new Text(this.getSession());
- }
-
- public Value asMutationValue() {
- return new Value(org.apache.accumulo.core.util.AddressUtil.toString(getLocation()).getBytes());
- }
-
- public InetSocketAddress getLocation() {
- return location;
- }
-
- public String getSession() {
- return session;
- }
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/88079cc3/server/base/src/main/java/org/apache/accumulo/server/master/state/TabletLocationState.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/master/state/TabletLocationState.java b/server/base/src/main/java/org/apache/accumulo/server/master/state/TabletLocationState.java
deleted file mode 100644
index bcfaead..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/master/state/TabletLocationState.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * 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.accumulo.server.master.state;
-
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Set;
-
-import org.apache.accumulo.core.data.KeyExtent;
-
-/**
- * When a tablet is assigned, we mark its future location. When the tablet is opened, we set its current location. A tablet should never have both a future and
- * current location.
- *
- * A tablet server is always associated with a unique session id. If the current tablet server has a different session, we know the location information is
- * out-of-date.
- */
-public class TabletLocationState {
-
- static public class BadLocationStateException extends Exception {
- private static final long serialVersionUID = 1L;
-
- BadLocationStateException(String msg) { super(msg); }
- }
-
- public TabletLocationState(KeyExtent extent, TServerInstance future, TServerInstance current, TServerInstance last, Collection<Collection<String>> walogs,
- boolean chopped) throws BadLocationStateException {
- this.extent = extent;
- this.future = future;
- this.current = current;
- this.last = last;
- if (walogs == null)
- walogs = Collections.emptyList();
- this.walogs = walogs;
- this.chopped = chopped;
- if (current != null && future != null) {
- throw new BadLocationStateException(extent + " is both assigned and hosted, which should never happen: " + this);
- }
- }
-
- final public KeyExtent extent;
- final public TServerInstance future;
- final public TServerInstance current;
- final public TServerInstance last;
- final public Collection<Collection<String>> walogs;
- final public boolean chopped;
-
- public String toString() {
- return extent + "@(" + future + "," + current + "," + last + ")" + (chopped ? " chopped" : "");
- }
-
- public TServerInstance getServer() {
- TServerInstance result = null;
- if (current != null) {
- result = current;
- } else if (future != null) {
- result = future;
- } else {
- result = last;
- }
- return result;
- }
-
- public TabletState getState(Set<TServerInstance> liveServers) {
- TServerInstance server = getServer();
- if (server == null)
- return TabletState.UNASSIGNED;
- if (server.equals(current) || server.equals(future)) {
- if (liveServers.contains(server))
- if (server.equals(future)) {
- return TabletState.ASSIGNED;
- } else {
- return TabletState.HOSTED;
- }
- else {
- return TabletState.ASSIGNED_TO_DEAD_SERVER;
- }
- }
- // server == last
- return TabletState.UNASSIGNED;
- }
-
-}