accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ktur...@apache.org
Subject [2/2] accumulo git commit: Merge branch '1.7' into 1.8
Date Fri, 27 Jan 2017 00:45:43 GMT
Merge branch '1.7' into 1.8


Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/bf5b6e0f
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/bf5b6e0f
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/bf5b6e0f

Branch: refs/heads/1.8
Commit: bf5b6e0fae4974bb77b6409c21c5770d815b991f
Parents: 8668c7f 7b9a11a
Author: Keith Turner <kturner@apache.org>
Authored: Thu Jan 26 19:24:14 2017 -0500
Committer: Keith Turner <kturner@apache.org>
Committed: Thu Jan 26 19:24:14 2017 -0500

----------------------------------------------------------------------
 .../accumulo/master/FateServiceHandler.java     |  15 +-
 .../master/tableOps/CancelCompactions.java      |  19 ++-
 .../master/tableOps/ChangeTableState.java       |  20 ++-
 .../accumulo/master/tableOps/CompactRange.java  |  30 ++--
 .../master/tableOps/CompactionDriver.java       |  14 +-
 .../accumulo/master/tableOps/DeleteTable.java   |  29 ++--
 .../accumulo/master/tableOps/RenameTable.java   |  19 ++-
 .../accumulo/master/tableOps/TableRangeOp.java  |  20 ++-
 .../master/tableOps/TableRangeOpWait.java       |   9 +-
 .../apache/accumulo/master/tableOps/Utils.java  |   9 +-
 .../functional/ConcurrentDeleteTableIT.java     | 167 ++++++++++++++++++-
 11 files changed, 262 insertions(+), 89 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/bf5b6e0f/server/master/src/main/java/org/apache/accumulo/master/FateServiceHandler.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/accumulo/blob/bf5b6e0f/server/master/src/main/java/org/apache/accumulo/master/tableOps/CompactRange.java
----------------------------------------------------------------------
diff --cc server/master/src/main/java/org/apache/accumulo/master/tableOps/CompactRange.java
index d7d5b14,e641479..f3d53a7
--- a/server/master/src/main/java/org/apache/accumulo/master/tableOps/CompactRange.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/tableOps/CompactRange.java
@@@ -24,11 -24,10 +24,10 @@@ import java.util.List
  import org.apache.accumulo.core.Constants;
  import org.apache.accumulo.core.client.IteratorSetting;
  import org.apache.accumulo.core.client.admin.CompactionStrategyConfig;
 +import org.apache.accumulo.core.client.impl.AcceptableThriftTableOperationException;
  import org.apache.accumulo.core.client.impl.CompactionStrategyConfigUtil;
- import org.apache.accumulo.core.client.impl.Tables;
  import org.apache.accumulo.core.client.impl.thrift.TableOperation;
  import org.apache.accumulo.core.client.impl.thrift.TableOperationExceptionType;
 -import org.apache.accumulo.core.client.impl.thrift.ThriftTableOperationException;
  import org.apache.accumulo.fate.Repo;
  import org.apache.accumulo.fate.zookeeper.IZooReaderWriter;
  import org.apache.accumulo.fate.zookeeper.IZooReaderWriter.Mutator;
@@@ -51,9 -51,14 +51,14 @@@ public class CompactRange extends Maste
    private byte[] endRow;
    private byte[] config;
  
-   public CompactRange(String tableId, byte[] startRow, byte[] endRow, List<IteratorSetting>
iterators, CompactionStrategyConfig compactionStrategy)
-       throws AcceptableThriftTableOperationException {
+   private String getNamespaceId(Master env) throws Exception {
+     return Utils.getNamespaceId(env.getInstance(), tableId, TableOperation.COMPACT, this.namespaceId);
+   }
+ 
+   public CompactRange(String namespaceId, String tableId, byte[] startRow, byte[] endRow,
List<IteratorSetting> iterators,
 -      CompactionStrategyConfig compactionStrategy) throws ThriftTableOperationException
{
++      CompactionStrategyConfig compactionStrategy) throws AcceptableThriftTableOperationException
{
  
+     requireNonNull(namespaceId, "Invalid argument: null namespaceId");
      requireNonNull(tableId, "Invalid argument: null tableId");
      requireNonNull(iterators, "Invalid argument: null iterator list");
      requireNonNull(compactionStrategy, "Invalid argument: null compactionStrategy");
@@@ -122,9 -127,9 +127,9 @@@
          }
        });
  
-       return new CompactionDriver(Long.parseLong(new String(cid, UTF_8).split(",")[0]),
tableId, startRow, endRow);
+       return new CompactionDriver(Long.parseLong(new String(cid, UTF_8).split(",")[0]),
getNamespaceId(env), tableId, startRow, endRow);
      } catch (NoNodeException nne) {
 -      throw new ThriftTableOperationException(tableId, null, TableOperation.COMPACT, TableOperationExceptionType.NOTFOUND,
null);
 +      throw new AcceptableThriftTableOperationException(tableId, null, TableOperation.COMPACT,
TableOperationExceptionType.NOTFOUND, null);
      }
  
    }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/bf5b6e0f/server/master/src/main/java/org/apache/accumulo/master/tableOps/CompactionDriver.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/accumulo/blob/bf5b6e0f/server/master/src/main/java/org/apache/accumulo/master/tableOps/RenameTable.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/accumulo/blob/bf5b6e0f/server/master/src/main/java/org/apache/accumulo/master/tableOps/TableRangeOp.java
----------------------------------------------------------------------
diff --cc server/master/src/main/java/org/apache/accumulo/master/tableOps/TableRangeOp.java
index e2e7018,64d08be..d91755d
--- a/server/master/src/main/java/org/apache/accumulo/master/tableOps/TableRangeOp.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/tableOps/TableRangeOp.java
@@@ -16,10 -16,9 +16,9 @@@
   */
  package org.apache.accumulo.master.tableOps;
  
 +import org.apache.accumulo.core.client.impl.AcceptableThriftTableOperationException;
- import org.apache.accumulo.core.client.impl.Tables;
  import org.apache.accumulo.core.client.impl.thrift.TableOperation;
  import org.apache.accumulo.core.client.impl.thrift.TableOperationExceptionType;
 -import org.apache.accumulo.core.client.impl.thrift.ThriftTableOperationException;
  import org.apache.accumulo.core.data.impl.KeyExtent;
  import org.apache.accumulo.core.metadata.RootTable;
  import org.apache.accumulo.core.util.TextUtil;
@@@ -42,15 -42,19 +42,19 @@@ public class TableRangeOp extends Maste
    private byte[] endRow;
    private Operation op;
  
-   @Override
-   public long isReady(long tid, Master environment) throws Exception {
-     String namespaceId = Tables.getNamespaceId(environment.getInstance(), tableId);
-     return Utils.reserveNamespace(namespaceId, tid, false, true, TableOperation.MERGE) +
Utils.reserveTable(tableId, tid, true, true, TableOperation.MERGE);
+   private String getNamespaceId(Master env) throws Exception {
+     return Utils.getNamespaceId(env.getInstance(), tableId, TableOperation.MERGE, this.namespaceId);
    }
  
-   public TableRangeOp(MergeInfo.Operation op, String tableId, Text startRow, Text endRow)
throws AcceptableThriftTableOperationException {
+   @Override
+   public long isReady(long tid, Master env) throws Exception {
+     return Utils.reserveNamespace(getNamespaceId(env), tid, false, true, TableOperation.MERGE)
+         + Utils.reserveTable(tableId, tid, true, true, TableOperation.MERGE);
+   }
  
 -  public TableRangeOp(MergeInfo.Operation op, String namespaceId, String tableId, Text startRow,
Text endRow) throws ThriftTableOperationException {
++  public TableRangeOp(MergeInfo.Operation op, String namespaceId, String tableId, Text startRow,
Text endRow) throws AcceptableThriftTableOperationException {
      this.tableId = tableId;
+     this.namespaceId = namespaceId;
      this.startRow = TextUtil.getBytes(startRow);
      this.endRow = TextUtil.getBytes(endRow);
      this.op = op;
@@@ -85,14 -90,14 +89,14 @@@
  
    @Override
    public void undo(long tid, Master env) throws Exception {
-     String namespaceId = Tables.getNamespaceId(env.getInstance(), tableId);
      // Not sure this is a good thing to do. The Master state engine should be the one to
remove it.
 -    Text tableIdText = new Text(tableId);
 -    MergeInfo mergeInfo = env.getMergeInfo(tableIdText);
 +    MergeInfo mergeInfo = env.getMergeInfo(tableId);
      if (mergeInfo.getState() != MergeState.NONE)
        log.info("removing merge information " + mergeInfo);
 -    env.clearMergeState(tableIdText);
 +    env.clearMergeState(tableId);
 +    Utils.unreserveNamespace(namespaceId, tid, false);
      Utils.unreserveTable(tableId, tid, true);
+     Utils.unreserveNamespace(getNamespaceId(env), tid, false);
    }
  
  }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/bf5b6e0f/server/master/src/main/java/org/apache/accumulo/master/tableOps/TableRangeOpWait.java
----------------------------------------------------------------------
diff --cc server/master/src/main/java/org/apache/accumulo/master/tableOps/TableRangeOpWait.java
index a7c82b1,5feb06d..1194c67
--- a/server/master/src/main/java/org/apache/accumulo/master/tableOps/TableRangeOpWait.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/tableOps/TableRangeOpWait.java
@@@ -57,12 -61,12 +59,11 @@@ class TableRangeOpWait extends MasterRe
  
    @Override
    public Repo<Master> call(long tid, Master master) throws Exception {
-     String namespaceId = Tables.getNamespaceId(master.getInstance(), tableId);
 -    Text tableIdText = new Text(tableId);
 -    MergeInfo mergeInfo = master.getMergeInfo(tableIdText);
 +    MergeInfo mergeInfo = master.getMergeInfo(tableId);
      log.info("removing merge information " + mergeInfo);
 -    master.clearMergeState(tableIdText);
 +    master.clearMergeState(tableId);
-     Utils.unreserveNamespace(namespaceId, tid, false);
      Utils.unreserveTable(tableId, tid, true);
+     Utils.unreserveNamespace(Utils.getNamespaceId(master.getInstance(), tableId, TableOperation.MERGE,
this.namespaceId), tid, false);
      return null;
    }
  

http://git-wip-us.apache.org/repos/asf/accumulo/blob/bf5b6e0f/server/master/src/main/java/org/apache/accumulo/master/tableOps/Utils.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/accumulo/blob/bf5b6e0f/test/src/main/java/org/apache/accumulo/test/functional/ConcurrentDeleteTableIT.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/functional/ConcurrentDeleteTableIT.java
index 3f7a305,0000000..5808804
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/ConcurrentDeleteTableIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/ConcurrentDeleteTableIT.java
@@@ -1,147 -1,0 +1,298 @@@
 +/*
 + * 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.test.functional;
 +
 +import java.util.ArrayList;
++import java.util.Collections;
 +import java.util.List;
++import java.util.Map;
 +import java.util.Random;
++import java.util.Set;
 +import java.util.TreeSet;
 +import java.util.concurrent.CountDownLatch;
 +import java.util.concurrent.ExecutorService;
 +import java.util.concurrent.Executors;
 +import java.util.concurrent.Future;
 +
 +import org.apache.accumulo.core.Constants;
 +import org.apache.accumulo.core.client.AccumuloException;
 +import org.apache.accumulo.core.client.AccumuloSecurityException;
 +import org.apache.accumulo.core.client.BatchWriter;
 +import org.apache.accumulo.core.client.BatchWriterConfig;
 +import org.apache.accumulo.core.client.Connector;
 +import org.apache.accumulo.core.client.Instance;
 +import org.apache.accumulo.core.client.MutationsRejectedException;
 +import org.apache.accumulo.core.client.TableNotFoundException;
++import org.apache.accumulo.core.client.TableOfflineException;
++import org.apache.accumulo.core.client.admin.CompactionConfig;
 +import org.apache.accumulo.core.conf.Property;
 +import org.apache.accumulo.core.data.Mutation;
 +import org.apache.accumulo.core.security.Authorizations;
 +import org.apache.accumulo.core.zookeeper.ZooUtil;
 +import org.apache.accumulo.fate.AdminUtil;
 +import org.apache.accumulo.fate.AdminUtil.FateStatus;
 +import org.apache.accumulo.fate.ZooStore;
 +import org.apache.accumulo.fate.zookeeper.IZooReaderWriter;
 +import org.apache.accumulo.harness.AccumuloClusterHarness;
 +import org.apache.accumulo.server.zookeeper.ZooReaderWriterFactory;
 +import org.apache.hadoop.io.Text;
 +import org.apache.zookeeper.KeeperException;
 +import org.junit.Assert;
 +import org.junit.Test;
 +
 +public class ConcurrentDeleteTableIT extends AccumuloClusterHarness {
 +
 +  @Test
 +  public void testConcurrentDeleteTablesOps() throws Exception {
 +    final Connector c = getConnector();
 +    String[] tables = getUniqueNames(2);
 +
-     TreeSet<Text> splits = new TreeSet<>();
- 
-     for (int i = 0; i < 1000; i++) {
-       Text split = new Text(String.format("%09x", i * 100000));
-       splits.add(split);
-     }
++    TreeSet<Text> splits = createSplits();
 +
 +    ExecutorService es = Executors.newFixedThreadPool(20);
 +
 +    int count = 0;
 +    for (final String table : tables) {
 +      c.tableOperations().create(table);
 +      c.tableOperations().addSplits(table, splits);
 +      writeData(c, table);
 +      if (count == 1) {
 +        c.tableOperations().flush(table, null, null, true);
 +      }
 +      count++;
 +
-       final CountDownLatch cdl = new CountDownLatch(20);
++      int numDeleteOps = 20;
++      final CountDownLatch cdl = new CountDownLatch(numDeleteOps);
 +
 +      List<Future<?>> futures = new ArrayList<>();
 +
-       for (int i = 0; i < 20; i++) {
++      for (int i = 0; i < numDeleteOps; i++) {
 +        Future<?> future = es.submit(new Runnable() {
 +
 +          @Override
 +          public void run() {
 +            try {
 +              cdl.countDown();
 +              cdl.await();
 +              c.tableOperations().delete(table);
 +            } catch (TableNotFoundException e) {
 +              // expected
 +            } catch (InterruptedException | AccumuloException | AccumuloSecurityException
e) {
 +              throw new RuntimeException(e);
 +            }
 +          }
 +        });
 +
 +        futures.add(future);
 +      }
 +
 +      for (Future<?> future : futures) {
 +        future.get();
 +      }
 +
 +      try {
 +        c.createScanner(table, Authorizations.EMPTY);
 +        Assert.fail("Expected table " + table + " to be gone.");
 +      } catch (TableNotFoundException tnfe) {
 +        // expected
 +      }
 +
 +      FateStatus fateStatus = getFateStatus();
 +
 +      // ensure there are no dangling locks... before ACCUMULO-4575 was fixed concurrent
delete tables could fail and leave dangling locks.
 +      Assert.assertEquals(0, fateStatus.getDanglingHeldLocks().size());
 +      Assert.assertEquals(0, fateStatus.getDanglingWaitingLocks().size());
 +    }
 +
 +    es.shutdown();
 +  }
 +
++  private TreeSet<Text> createSplits() {
++    TreeSet<Text> splits = new TreeSet<>();
++
++    for (int i = 0; i < 1000; i++) {
++      Text split = new Text(String.format("%09x", i * 100000));
++      splits.add(split);
++    }
++    return splits;
++  }
++
++  private static abstract class DelayedTableOp implements Runnable {
++    private CountDownLatch cdl;
++
++    DelayedTableOp(CountDownLatch cdl) {
++      this.cdl = cdl;
++    }
++
++    public void run() {
++      try {
++        cdl.countDown();
++        cdl.await();
++        Thread.sleep(10);
++        doTableOp();
++      } catch (TableNotFoundException e) {
++        // expected
++      } catch (RuntimeException e) {
++        throw e;
++      } catch (Exception e) {
++        throw new RuntimeException(e);
++      }
++    }
++
++    protected abstract void doTableOp() throws Exception;
++  }
++
++  @Test
++  public void testConcurrentFateOpsWithDelete() throws Exception {
++    final Connector c = getConnector();
++    String[] tables = getUniqueNames(2);
++
++    TreeSet<Text> splits = createSplits();
++
++    int numOperations = 8;
++
++    ExecutorService es = Executors.newFixedThreadPool(numOperations);
++
++    int count = 0;
++    for (final String table : tables) {
++      c.tableOperations().create(table);
++      c.tableOperations().addSplits(table, splits);
++      writeData(c, table);
++      if (count == 1) {
++        c.tableOperations().flush(table, null, null, true);
++      }
++      count++;
++
++      // increment this for each test
++      final CountDownLatch cdl = new CountDownLatch(numOperations);
++
++      List<Future<?>> futures = new ArrayList<>();
++
++      futures.add(es.submit(new Runnable() {
++        @Override
++        public void run() {
++          try {
++            cdl.countDown();
++            cdl.await();
++            c.tableOperations().delete(table);
++          } catch (TableNotFoundException | TableOfflineException e) {
++            // expected
++          } catch (InterruptedException | AccumuloException | AccumuloSecurityException
e) {
++            throw new RuntimeException(e);
++          }
++        }
++      }));
++
++      futures.add(es.submit(new DelayedTableOp(cdl) {
++        @Override
++        protected void doTableOp() throws Exception {
++          c.tableOperations().compact(table, new CompactionConfig());
++        }
++      }));
++
++      futures.add(es.submit(new DelayedTableOp(cdl) {
++        @Override
++        protected void doTableOp() throws Exception {
++          c.tableOperations().merge(table, null, null);
++        }
++      }));
++
++      futures.add(es.submit(new DelayedTableOp(cdl) {
++        @Override
++        protected void doTableOp() throws Exception {
++          Map<String,String> m = Collections.emptyMap();
++          Set<String> s = Collections.emptySet();
++          c.tableOperations().clone(table, table + "_clone", true, m, s);
++        }
++      }));
++
++      futures.add(es.submit(new DelayedTableOp(cdl) {
++        @Override
++        protected void doTableOp() throws Exception {
++          c.tableOperations().deleteRows(table, null, null);
++        }
++      }));
++
++      futures.add(es.submit(new DelayedTableOp(cdl) {
++        @Override
++        protected void doTableOp() throws Exception {
++          c.tableOperations().cancelCompaction(table);
++        }
++      }));
++
++      futures.add(es.submit(new DelayedTableOp(cdl) {
++        @Override
++        protected void doTableOp() throws Exception {
++          c.tableOperations().rename(table, table + "_renamed");
++        }
++      }));
++
++      futures.add(es.submit(new DelayedTableOp(cdl) {
++        @Override
++        protected void doTableOp() throws Exception {
++          c.tableOperations().offline(table);
++        }
++      }));
++
++      Assert.assertEquals(numOperations, futures.size());
++
++      for (Future<?> future : futures) {
++        future.get();
++      }
++
++      try {
++        c.createScanner(table, Authorizations.EMPTY);
++        Assert.fail("Expected table " + table + " to be gone.");
++      } catch (TableNotFoundException tnfe) {
++        // expected
++      }
++
++      FateStatus fateStatus = getFateStatus();
++
++      // ensure there are no dangling locks... before ACCUMULO-4575 was fixed concurrent
delete tables could fail and leave dangling locks.
++      Assert.assertEquals(0, fateStatus.getDanglingHeldLocks().size());
++      Assert.assertEquals(0, fateStatus.getDanglingWaitingLocks().size());
++    }
++
++    es.shutdown();
++  }
++
 +  private FateStatus getFateStatus() throws KeeperException, InterruptedException {
 +    Instance instance = getConnector().getInstance();
 +    AdminUtil<String> admin = new AdminUtil<>(false);
 +    String secret = getCluster().getSiteConfiguration().get(Property.INSTANCE_SECRET);
 +    IZooReaderWriter zk = new ZooReaderWriterFactory().getZooReaderWriter(instance.getZooKeepers(),
instance.getZooKeepersSessionTimeOut(), secret);
 +    ZooStore<String> zs = new ZooStore<String>(ZooUtil.getRoot(instance) + Constants.ZFATE,
zk);
 +    FateStatus fateStatus = admin.getStatus(zs, zk, ZooUtil.getRoot(instance) + Constants.ZTABLE_LOCKS,
null, null);
 +    return fateStatus;
 +  }
 +
 +  private void writeData(Connector c, String table) throws TableNotFoundException, MutationsRejectedException
{
 +    BatchWriter bw = c.createBatchWriter(table, new BatchWriterConfig());
 +    try {
 +      Random rand = new Random();
 +      for (int i = 0; i < 1000; i++) {
 +        Mutation m = new Mutation(String.format("%09x", rand.nextInt(100000 * 1000)));
 +        m.put("m", "order", "" + i);
 +        bw.addMutation(m);
 +      }
 +    } finally {
 +      bw.close();
 +    }
 +  }
 +}


Mime
View raw message