From 0097786b5892f4ae858ef00abf056c8fb8c342b4 Mon Sep 17 00:00:00 2001 From: Ashish Singhi Date: Thu, 25 Sep 2014 16:26:30 +0530 Subject: [PATCH] HBASE-12096 --- .../ZKSplitLogManagerCoordination.java | 40 +++++++++++++++------- .../coordination/ZkSplitLogWorkerCoordination.java | 7 ++-- .../hbase/coprocessor/AggregateImplementation.java | 29 ++++++++++------ 3 files changed, 49 insertions(+), 27 deletions(-) diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/coordination/ZKSplitLogManagerCoordination.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/coordination/ZKSplitLogManagerCoordination.java index b7974f7..de748fe 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/coordination/ZKSplitLogManagerCoordination.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/coordination/ZKSplitLogManagerCoordination.java @@ -152,8 +152,9 @@ public class ZKSplitLogManagerCoordination extends ZooKeeperListener implements try { List tasks = ZKUtil.listChildrenNoWatch(watcher, watcher.splitLogZNode); if (tasks != null) { - for (String t : tasks) { - if (!ZKSplitLog.isRescanNode(watcher, t)) { + int listSize = tasks.size(); + for (int i = 0; i < listSize; i++) { + if (!ZKSplitLog.isRescanNode(watcher, tasks.get(i))) { count++; } } @@ -302,8 +303,9 @@ public class ZKSplitLogManagerCoordination extends ZooKeeperListener implements try { List tasks = ZKUtil.listChildrenNoWatch(watcher, watcher.splitLogZNode); if (tasks != null) { - for (String t : tasks) { - if (!ZKSplitLog.isRescanNode(watcher, t)) { + int listSize = tasks.size(); + for (int i = 0; i < listSize; i++) { + if (!ZKSplitLog.isRescanNode(watcher, tasks.get(i))) { count++; } } @@ -319,7 +321,9 @@ public class ZKSplitLogManagerCoordination extends ZooKeeperListener implements // remove recovering regions which doesn't have any RS associated with it List regions = ZKUtil.listChildrenNoWatch(watcher, watcher.recoveringRegionsZNode); if (regions != null) { - for (String region : regions) { + int listSize = regions.size(); + for (int i = 0; i < listSize; i++) { + String region = regions.get(i); if (isMetaRecovery != null) { if ((isMetaRecovery && !region.equalsIgnoreCase(metaEncodeRegionName)) || (!isMetaRecovery && region.equalsIgnoreCase(metaEncodeRegionName))) { @@ -337,7 +341,9 @@ public class ZKSplitLogManagerCoordination extends ZooKeeperListener implements if (recoveredServerNameSet.containsAll(failedServers)) { ZKUtil.deleteNodeRecursively(watcher, nodePath); } else { - for (String failedServer : failedServers) { + listSize = failedServers.size(); + for (int j = 0; j < listSize; j++) { + String failedServer = failedServers.get(j); if (recoveredServerNameSet.contains(failedServer)) { String tmpPath = ZKUtil.joinZNode(nodePath, failedServer); ZKUtil.deleteNode(watcher, tmpPath); @@ -576,7 +582,9 @@ public class ZKSplitLogManagerCoordination extends ZooKeeperListener implements return; } int rescan_nodes = 0; - for (String path : orphans) { + int listSize = orphans.size(); + for (int i = 0; i < listSize; i++) { + String path = orphans.get(i); String nodepath = ZKUtil.joinZNode(watcher.splitLogZNode, path); if (ZKSplitLog.isRescanNode(watcher, nodepath)) { rescan_nodes++; @@ -682,7 +690,9 @@ public class ZKSplitLogManagerCoordination extends ZooKeeperListener implements try { List tasks = ZKUtil.listChildrenNoWatch(watcher, watcher.splitLogZNode); if (tasks != null) { - for (String t : tasks) { + int listSize = tasks.size(); + for (int i = 0; i < listSize; i++) { + String t = tasks.get(i); byte[] data; try { data = ZKUtil.getData(this.watcher, ZKUtil.joinZNode(watcher.splitLogZNode, t)); @@ -714,16 +724,18 @@ public class ZKSplitLogManagerCoordination extends ZooKeeperListener implements // remove recovering regions which doesn't have any RS associated with it List regions = ZKUtil.listChildrenNoWatch(watcher, watcher.recoveringRegionsZNode); if (regions != null) { - for (String region : regions) { - String nodePath = ZKUtil.joinZNode(watcher.recoveringRegionsZNode, region); + int listSize = regions.size(); + for (int i = 0; i < listSize; i++) { + String nodePath = ZKUtil.joinZNode(watcher.recoveringRegionsZNode, regions.get(i)); List regionFailedServers = ZKUtil.listChildrenNoWatch(watcher, nodePath); if (regionFailedServers == null || regionFailedServers.isEmpty()) { ZKUtil.deleteNode(watcher, nodePath); continue; } boolean needMoreRecovery = false; - for (String tmpFailedServer : regionFailedServers) { - if (knownFailedServers.contains(tmpFailedServer)) { + listSize = regionFailedServers.size(); + for (i = 0; i < listSize; i++) { + if (knownFailedServers.contains(regionFailedServers.get(i))) { needMoreRecovery = true; break; } @@ -787,7 +799,9 @@ public class ZKSplitLogManagerCoordination extends ZooKeeperListener implements hasSplitLogTask = true; if (isForInitialization) { // during initialization, try to get recovery mode from splitlogtask - for (String task : tasks) { + int listSize = tasks.size(); + for (int i = 0; i < listSize; i++) { + String task = tasks.get(i); try { byte[] data = ZKUtil.getData(this.watcher, ZKUtil.joinZNode(watcher.splitLogZNode, task)); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/coordination/ZkSplitLogWorkerCoordination.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/coordination/ZkSplitLogWorkerCoordination.java index ee37259..13e7ee9 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/coordination/ZkSplitLogWorkerCoordination.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/coordination/ZkSplitLogWorkerCoordination.java @@ -450,9 +450,10 @@ public class ZkSplitLogWorkerCoordination extends ZooKeeperListener implements // Make a local copy to prevent ConcurrentModificationException when other threads // modify recoveringRegions List tmpCopy = new ArrayList(recoveringRegions.keySet()); - for (String region : tmpCopy) { - String nodePath = - ZKUtil.joinZNode(watcher.recoveringRegionsZNode, region); + int listSize = tmpCopy.size(); + for (int i = 0; i < listSize; i++) { + String region = tmpCopy.get(i); + String nodePath = ZKUtil.joinZNode(watcher.recoveringRegionsZNode, region); try { if (ZKUtil.checkExists(watcher, nodePath) == -1) { HRegion r = recoveringRegions.remove(region); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/AggregateImplementation.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/AggregateImplementation.java index 7408c3b..f361022 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/AggregateImplementation.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/AggregateImplementation.java @@ -92,8 +92,9 @@ extends AggregateService implements CoprocessorService, Coprocessor { boolean hasMoreRows = false; do { hasMoreRows = scanner.next(results); - for (Cell kv : results) { - temp = ci.getValue(colFamily, qualifier, kv); + int listSize = results.size(); + for (int i = 0; i < listSize; i++) { + temp = ci.getValue(colFamily, qualifier, results.get(i)); max = (max == null || (temp != null && ci.compare(temp, max) > 0)) ? temp : max; } results.clear(); @@ -145,8 +146,9 @@ extends AggregateService implements CoprocessorService, Coprocessor { boolean hasMoreRows = false; do { hasMoreRows = scanner.next(results); - for (Cell kv : results) { - temp = ci.getValue(colFamily, qualifier, kv); + int listSize = results.size(); + for (int i = 0; i < listSize; i++) { + temp = ci.getValue(colFamily, qualifier, results.get(i)); min = (min == null || (temp != null && ci.compare(temp, min) < 0)) ? temp : min; } results.clear(); @@ -198,8 +200,9 @@ extends AggregateService implements CoprocessorService, Coprocessor { boolean hasMoreRows = false; do { hasMoreRows = scanner.next(results); - for (Cell kv : results) { - temp = ci.getValue(colFamily, qualifier, kv); + int listSize = results.size(); + for (int i = 0; i < listSize; i++) { + temp = ci.getValue(colFamily, qualifier, results.get(i)); if (temp != null) sumVal = ci.add(sumVal, ci.castToReturnType(temp)); } @@ -310,9 +313,10 @@ extends AggregateService implements CoprocessorService, Coprocessor { do { results.clear(); hasMoreRows = scanner.next(results); - for (Cell kv : results) { + int listSize = results.size(); + for (int i = 0; i < listSize; i++) { sumVal = ci.add(sumVal, ci.castToReturnType(ci.getValue(colFamily, - qualifier, kv))); + qualifier, results.get(i)))); } rowCountVal++; } while (hasMoreRows); @@ -370,9 +374,10 @@ extends AggregateService implements CoprocessorService, Coprocessor { do { tempVal = null; hasMoreRows = scanner.next(results); - for (Cell kv : results) { + int listSize = results.size(); + for (int i = 0; i < listSize; i++) { tempVal = ci.add(tempVal, ci.castToReturnType(ci.getValue(colFamily, - qualifier, kv))); + qualifier, results.get(i)))); } results.clear(); sumVal = ci.add(sumVal, tempVal); @@ -436,7 +441,9 @@ extends AggregateService implements CoprocessorService, Coprocessor { tempVal = null; tempWeight = null; hasMoreRows = scanner.next(results); - for (Cell kv : results) { + int listSize = results.size(); + for (int i = 0; i < listSize; i++) { + Cell kv = results.get(i); tempVal = ci.add(tempVal, ci.castToReturnType(ci.getValue(colFamily, valQualifier, kv))); if (weightQualifier != null) { -- 1.9.2.msysgit.0