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 2018/03/14 20:42:10 UTC
[accumulo] branch 1.8 updated (d2908b5 -> 1335249)
This is an automated email from the ASF dual-hosted git repository.
ctubbsii pushed a change to branch 1.8
in repository https://gitbox.apache.org/repos/asf/accumulo.git.
from d2908b5 Merge branch '1.7' into 1.8
add cf0dd9c ACCUMULO-4847 Fix Retry utility's API (#399)
new 1335249 Merge branch '1.7' into 1.8
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../core/client/impl/TableOperationsImpl.java | 7 +-
.../java/org/apache/accumulo/fate/util/Retry.java | 357 +++++++++++++++++++++
.../org/apache/accumulo/fate/zookeeper/Retry.java | 213 ------------
.../accumulo/fate/zookeeper/RetryFactory.java | 74 -----
.../apache/accumulo/fate/zookeeper/ZooReader.java | 20 +-
.../accumulo/fate/zookeeper/ZooReaderWriter.java | 7 +-
.../apache/accumulo/fate/zookeeper/ZooUtil.java | 30 +-
.../org/apache/accumulo/fate/util/RetryTest.java | 294 +++++++++++++++++
.../accumulo/fate/zookeeper/RetryFactoryTest.java | 53 ---
.../apache/accumulo/fate/zookeeper/RetryTest.java | 180 -----------
.../fate/zookeeper/ZooReaderWriterTest.java | 4 +-
.../org/apache/accumulo/tserver/TabletServer.java | 15 +-
.../accumulo/tserver/log/TabletServerLogger.java | 10 +-
13 files changed, 706 insertions(+), 558 deletions(-)
create mode 100644 fate/src/main/java/org/apache/accumulo/fate/util/Retry.java
delete mode 100644 fate/src/main/java/org/apache/accumulo/fate/zookeeper/Retry.java
delete mode 100644 fate/src/main/java/org/apache/accumulo/fate/zookeeper/RetryFactory.java
create mode 100644 fate/src/test/java/org/apache/accumulo/fate/util/RetryTest.java
delete mode 100644 fate/src/test/java/org/apache/accumulo/fate/zookeeper/RetryFactoryTest.java
delete mode 100644 fate/src/test/java/org/apache/accumulo/fate/zookeeper/RetryTest.java
--
To stop receiving notification emails like this one, please contact
ctubbsii@apache.org.
[accumulo] 01/01: Merge branch '1.7' into 1.8
Posted by ct...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ctubbsii pushed a commit to branch 1.8
in repository https://gitbox.apache.org/repos/asf/accumulo.git
commit 13352490b599001021e1e9e817a051f5ffa42561
Merge: d2908b5 cf0dd9c
Author: Christopher Tubbs <ct...@apache.org>
AuthorDate: Wed Mar 14 16:21:04 2018 -0400
Merge branch '1.7' into 1.8
.../core/client/impl/TableOperationsImpl.java | 7 +-
.../java/org/apache/accumulo/fate/util/Retry.java | 357 +++++++++++++++++++++
.../org/apache/accumulo/fate/zookeeper/Retry.java | 213 ------------
.../accumulo/fate/zookeeper/RetryFactory.java | 74 -----
.../apache/accumulo/fate/zookeeper/ZooReader.java | 20 +-
.../accumulo/fate/zookeeper/ZooReaderWriter.java | 7 +-
.../apache/accumulo/fate/zookeeper/ZooUtil.java | 30 +-
.../org/apache/accumulo/fate/util/RetryTest.java | 294 +++++++++++++++++
.../accumulo/fate/zookeeper/RetryFactoryTest.java | 53 ---
.../apache/accumulo/fate/zookeeper/RetryTest.java | 180 -----------
.../fate/zookeeper/ZooReaderWriterTest.java | 4 +-
.../org/apache/accumulo/tserver/TabletServer.java | 15 +-
.../accumulo/tserver/log/TabletServerLogger.java | 10 +-
13 files changed, 706 insertions(+), 558 deletions(-)
diff --cc core/src/main/java/org/apache/accumulo/core/client/impl/TableOperationsImpl.java
index be4999f,d976c4d..b3de313
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/TableOperationsImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/TableOperationsImpl.java
@@@ -18,11 -18,8 +18,13 @@@ package org.apache.accumulo.core.client
import static com.google.common.base.Preconditions.checkArgument;
import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Objects.requireNonNull;
++import static java.util.concurrent.TimeUnit.MILLISECONDS;
++import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.accumulo.fate.util.UtilWaitThread.sleepUninterruptibly;
import java.io.BufferedReader;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.ByteBuffer;
@@@ -113,8 -105,8 +115,8 @@@ import org.apache.accumulo.core.util.Na
import org.apache.accumulo.core.util.OpTimer;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.core.util.TextUtil;
-import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.core.volume.VolumeConfiguration;
- import org.apache.accumulo.fate.zookeeper.Retry;
++import org.apache.accumulo.fate.util.Retry;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@@ -1504,148 -1466,4 +1506,149 @@@ public class TableOperationsImpl extend
}
}
+ private void clearSamplerOptions(String tableName) throws AccumuloException, TableNotFoundException, AccumuloSecurityException {
+ String prefix = Property.TABLE_SAMPLER_OPTS.getKey();
+ for (Entry<String,String> entry : getProperties(tableName)) {
+ String property = entry.getKey();
+ if (property.startsWith(prefix)) {
+ removeProperty(tableName, property);
+ }
+ }
+ }
+
+ @Override
+ public void setSamplerConfiguration(String tableName, SamplerConfiguration samplerConfiguration) throws AccumuloException, TableNotFoundException,
+ AccumuloSecurityException {
+ clearSamplerOptions(tableName);
+
+ List<Pair<String,String>> props = new SamplerConfigurationImpl(samplerConfiguration).toTableProperties();
+ for (Pair<String,String> pair : props) {
+ setProperty(tableName, pair.getFirst(), pair.getSecond());
+ }
+ }
+
+ @Override
+ public void clearSamplerConfiguration(String tableName) throws AccumuloException, TableNotFoundException, AccumuloSecurityException {
+ removeProperty(tableName, Property.TABLE_SAMPLER.getKey());
+ clearSamplerOptions(tableName);
+ }
+
+ @Override
+ public SamplerConfiguration getSamplerConfiguration(String tableName) throws TableNotFoundException, AccumuloException {
+ AccumuloConfiguration conf = new ConfigurationCopy(this.getProperties(tableName));
+ SamplerConfigurationImpl sci = SamplerConfigurationImpl.newSamplerConfig(conf);
+ if (sci == null) {
+ return null;
+ }
+ return sci.toSamplerConfiguration();
+ }
+
+ private static class LoctionsImpl implements Locations {
+
+ private Map<Range,List<TabletId>> groupedByRanges;
+ private Map<TabletId,List<Range>> groupedByTablets;
+ private Map<TabletId,String> tabletLocations;
+
+ public LoctionsImpl(Map<String,Map<KeyExtent,List<Range>>> binnedRanges) {
+ groupedByTablets = new HashMap<>();
+ groupedByRanges = null;
+ tabletLocations = new HashMap<>();
+
+ for (Entry<String,Map<KeyExtent,List<Range>>> entry : binnedRanges.entrySet()) {
+ String location = entry.getKey();
+
+ for (Entry<KeyExtent,List<Range>> entry2 : entry.getValue().entrySet()) {
+ TabletIdImpl tabletId = new TabletIdImpl(entry2.getKey());
+ tabletLocations.put(tabletId, location);
+ List<Range> prev = groupedByTablets.put(tabletId, Collections.unmodifiableList(entry2.getValue()));
+ if (prev != null) {
+ throw new RuntimeException("Unexpected : tablet at multiple locations : " + location + " " + tabletId);
+ }
+ }
+ }
+
+ groupedByTablets = Collections.unmodifiableMap(groupedByTablets);
+ }
+
+ @Override
+ public String getTabletLocation(TabletId tabletId) {
+ return tabletLocations.get(tabletId);
+ }
+
+ @Override
+ public Map<Range,List<TabletId>> groupByRange() {
+ if (groupedByRanges == null) {
+ Map<Range,List<TabletId>> tmp = new HashMap<>();
+
+ for (Entry<TabletId,List<Range>> entry : groupedByTablets.entrySet()) {
+ for (Range range : entry.getValue()) {
+ List<TabletId> tablets = tmp.get(range);
+ if (tablets == null) {
+ tablets = new ArrayList<>();
+ tmp.put(range, tablets);
+ }
+
+ tablets.add(entry.getKey());
+ }
+ }
+
+ Map<Range,List<TabletId>> tmp2 = new HashMap<>();
+ for (Entry<Range,List<TabletId>> entry : tmp.entrySet()) {
+ tmp2.put(entry.getKey(), Collections.unmodifiableList(entry.getValue()));
+ }
+
+ groupedByRanges = Collections.unmodifiableMap(tmp2);
+ }
+
+ return groupedByRanges;
+ }
+
+ @Override
+ public Map<TabletId,List<Range>> groupByTablet() {
+ return groupedByTablets;
+ }
+ }
+
+ @Override
+ public Locations locate(String tableName, Collection<Range> ranges) throws AccumuloException, AccumuloSecurityException, TableNotFoundException {
+ requireNonNull(tableName, "tableName must be non null");
+ requireNonNull(ranges, "ranges must be non null");
+
+ String tableId = Tables.getTableId(context.getInstance(), tableName);
+ TabletLocator locator = TabletLocator.getLocator(context, tableId);
+
+ List<Range> rangeList = null;
+ if (ranges instanceof List) {
+ rangeList = (List<Range>) ranges;
+ } else {
+ rangeList = new ArrayList<>(ranges);
+ }
+
+ Map<String,Map<KeyExtent,List<Range>>> binnedRanges = new HashMap<>();
+
+ locator.invalidateCache();
+
- Retry retry = new Retry(Long.MAX_VALUE, 100, 100, 2000);
++ Retry retry = Retry.builder().infiniteRetries().retryAfter(100, MILLISECONDS).incrementBy(100, MILLISECONDS).maxWait(2, SECONDS)
++ .logInterval(3, TimeUnit.MINUTES).createRetry();
+
+ while (!locator.binRanges(context, rangeList, binnedRanges).isEmpty()) {
+
+ if (!Tables.exists(context.getInstance(), tableId))
+ throw new TableNotFoundException(tableId, tableName, null);
+ if (Tables.getTableState(context.getInstance(), tableId) == TableState.OFFLINE)
+ throw new TableOfflineException(context.getInstance(), tableId);
+
+ binnedRanges.clear();
+
+ try {
+ retry.waitForNextAttempt();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+
+ locator.invalidateCache();
+ }
+
+ return new LoctionsImpl(binnedRanges);
+ }
}
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index cf555ac,1a5787b..a4012db
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@@ -147,12 -137,12 +147,13 @@@ import org.apache.accumulo.core.util.Pa
import org.apache.accumulo.core.util.ServerServices;
import org.apache.accumulo.core.util.ServerServices.Service;
import org.apache.accumulo.core.util.SimpleThreadPool;
-import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.accumulo.core.util.ratelimit.RateLimiter;
+import org.apache.accumulo.core.util.ratelimit.SharedRateLimiterFactory;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.fate.util.LoggingRunnable;
+ import org.apache.accumulo.fate.util.Retry;
+ import org.apache.accumulo.fate.util.Retry.RetryFactory;
import org.apache.accumulo.fate.zookeeper.IZooReaderWriter;
- import org.apache.accumulo.fate.zookeeper.RetryFactory;
import org.apache.accumulo.fate.zookeeper.ZooLock.LockLossReason;
import org.apache.accumulo.fate.zookeeper.ZooLock.LockWatcher;
import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeExistsPolicy;
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java
index 92efb9c,abf5645..a2e90ac
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java
@@@ -39,10 -36,8 +39,10 @@@ import org.apache.accumulo.core.data.Mu
import org.apache.accumulo.core.data.impl.KeyExtent;
import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationConfigurationUtil;
+import org.apache.accumulo.core.util.SimpleThreadPool;
+import org.apache.accumulo.fate.util.LoggingRunnable;
- import org.apache.accumulo.fate.zookeeper.Retry;
- import org.apache.accumulo.fate.zookeeper.RetryFactory;
+ import org.apache.accumulo.fate.util.Retry;
+ import org.apache.accumulo.fate.util.Retry.RetryFactory;
import org.apache.accumulo.server.conf.TableConfiguration;
import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.replication.StatusUtil;
@@@ -213,29 -201,21 +213,29 @@@ public class TabletServerLogger
}
try {
- DfsLogger alog = new DfsLogger(tserver.getServerConfig(), syncCounter, flushCounter);
- alog.open(tserver.getClientAddressString());
- loggers.add(alog);
- logSetId.incrementAndGet();
-
- // When we successfully create a WAL, make sure to reset the Retry.
- if (null != createRetry) {
- createRetry = null;
+ startLogMaker();
+ Object next = nextLog.take();
+ if (next instanceof Exception) {
+ throw (Exception) next;
}
+ if (next instanceof DfsLogger) {
+ currentLog = (DfsLogger) next;
+ logId.incrementAndGet();
+ log.info("Using next log " + currentLog.getFileName());
+
+ // When we successfully create a WAL, make sure to reset the Retry.
+ if (null != createRetry) {
+ createRetry = null;
+ }
- this.createTime = System.currentTimeMillis();
- return;
+ this.createTime = System.currentTimeMillis();
+ return;
+ } else {
+ throw new RuntimeException("Error: unexpected type seen: " + next);
+ }
} catch (Exception t) {
if (null == createRetry) {
- createRetry = createRetryFactory.create();
+ createRetry = createRetryFactory.createRetry();
}
// We have more retries or we exceeded the maximum number of accepted failures
--
To stop receiving notification emails like this one, please contact
ctubbsii@apache.org.