accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From els...@apache.org
Subject [47/50] [abbrv] git commit: ACCUMULO-2575 Fix bug in GC where files were being prematurely deleted and add tests to catch
Date Fri, 09 May 2014 15:29:25 GMT
ACCUMULO-2575 Fix bug in GC where files were being prematurely deleted and add tests to catch

After realizing the deadlock potential when not having tservers write the initial
replication entry to the metadata table, the garbage collection code was not updated
for the shift in data layout. On a clean system, when first beginning replication,
the replication table will not yet exist, but replication entries will exist in the
metadata table.

The GC code still assuming that the lack of a replication table implied that
there was no data pending replication and short-circuited early. We need to check
the metadata table always, and, when present, also the replication table.


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

Branch: refs/heads/ACCUMULO-378
Commit: 53fc90fcc783a37df666915cec45f3da3a290c7f
Parents: 440d7fc
Author: Josh Elser <elserj@apache.org>
Authored: Thu May 8 16:16:33 2014 -0400
Committer: Josh Elser <elserj@apache.org>
Committed: Thu May 8 16:16:33 2014 -0400

----------------------------------------------------------------------
 .../gc/GarbageCollectWriteAheadLogs.java        | 70 ++++++++++++++------
 .../gc/GarbageCollectWriteAheadLogsTest.java    | 68 ++++++++++++++++++-
 .../replication/ReplicationWorkAssigner.java    |  2 +
 .../test/replication/ReplicationIT.java         |  6 +-
 .../test/replication/ReplicationWithGCIT.java   | 60 ++++-------------
 test/src/test/resources/log4j.properties        |  2 +
 6 files changed, 135 insertions(+), 73 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/53fc90fc/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
----------------------------------------------------------------------
diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
index 6e34e43..950bc12 100644
--- a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
@@ -66,16 +66,18 @@ import org.apache.accumulo.trace.instrument.Trace;
 import org.apache.accumulo.trace.instrument.Tracer;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
-import org.apache.log4j.Logger;
 import org.apache.thrift.TException;
 import org.apache.zookeeper.KeeperException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.google.common.collect.Iterables;
 import com.google.common.net.HostAndPort;
 import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.TextFormat;
 
 public class GarbageCollectWriteAheadLogs {
-  private static final Logger log = Logger.getLogger(GarbageCollectWriteAheadLogs.class);
+  private static final Logger log = LoggerFactory.getLogger(GarbageCollectWriteAheadLogs.class);
   
   private final Instance instance;
   private final VolumeManager fs;
@@ -191,7 +193,7 @@ public class GarbageCollectWriteAheadLogs {
     } catch (KeeperException.NoNodeException ex) {
       return false;
     } catch (Exception ex) {
-      log.debug(ex, ex);
+      log.debug(ex.toString(), ex);
       return true;
     }
   }
@@ -349,13 +351,15 @@ public class GarbageCollectWriteAheadLogs {
       Entry<String,Path> wal = walIter.next();
       String fullPath = wal.getValue().toString();
       if (neededByReplication(conn, fullPath)) {
-        log.debug("Removing WAL from candidate deletion as it is still needed for replication:
" + fullPath);
+        log.debug("Removing WAL from candidate deletion as it is still needed for replication:
{} ", fullPath);
         // If we haven't already removed it, check to see if this WAL is
         // "in use" by replication (needed for replication purposes)
         status.currentLog.inUse++;
 
         walIter.remove();
         sortedWALogs.remove(wal.getKey());
+      } else {
+        log.debug("WAL not needed for replication {}", fullPath);
       }
       count++;
     }
@@ -370,13 +374,25 @@ public class GarbageCollectWriteAheadLogs {
    */
   protected boolean neededByReplication(Connector conn, String wal) {
     log.info("Checking replication table for " + wal);
-    Iterable<Entry<Key,Value>> iter;
-    try {
-      iter = getReplicationStatusForFile(conn, wal);
-    } catch (TableNotFoundException e) {
-      log.trace("Replication table was not found");
-      return false;
-    }
+
+    // try {
+    // log.info("Current state of Metadata table");
+    // for (Entry<Key,Value> entry : conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY))
{
+    // log.info(entry.getKey().toStringNoTruncate() + "=" + TextFormat.shortDebugString(Status.parseFrom(entry.getValue().get())));
+    // }
+    // } catch (Exception e) {
+    // log.error("Could not read metadata table");
+    // }
+    // try {
+    // log.info("Current state of replication table");
+    // for (Entry<Key,Value> entry : conn.createScanner(ReplicationTable.NAME, Authorizations.EMPTY))
{
+    // log.info(entry.getKey().toStringNoTruncate() + "=" + TextFormat.shortDebugString(Status.parseFrom(entry.getValue().get())));
+    // }
+    // } catch (Exception e) {
+    // log.error("Could not read replication table");
+    // }
+    
+    Iterable<Entry<Key,Value>> iter = getReplicationStatusForFile(conn, wal);
 
     // TODO Push down this filter to the tserver to only return records
     // that are not completely replicated and convert this loop into a
@@ -384,6 +400,7 @@ public class GarbageCollectWriteAheadLogs {
     for (Entry<Key,Value> entry : iter) {
       try {
         Status status = Status.parseFrom(entry.getValue().get());
+        log.info("Checking if {} is safe for removal with {}", wal, TextFormat.shortDebugString(status));
         if (!StatusUtil.isSafeForRemoval(status)) {
           return true;
         }
@@ -395,21 +412,34 @@ public class GarbageCollectWriteAheadLogs {
     return false;
   }
 
-  protected Iterable<Entry<Key,Value>> getReplicationStatusForFile(Connector
conn, String wal) throws TableNotFoundException {
-    Scanner replScanner = ReplicationTable.getScanner(conn);
-
-    // Scan only the Status records
-    StatusSection.limit(replScanner);
-    // Only look for this specific WAL
-    replScanner.setRange(Range.exact(wal));
+  protected Iterable<Entry<Key,Value>> getReplicationStatusForFile(Connector
conn, String wal) {
+    Scanner metaScanner;
+    try {
+      metaScanner = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
+    } catch (TableNotFoundException e) {
+      throw new RuntimeException(e);
+    }
 
-    Scanner metaScanner = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
     // Need to add in the replication section prefix
     metaScanner.setRange(Range.exact(ReplicationSection.getRowPrefix() + wal));
     // Limit the column family to be sure
     metaScanner.fetchColumnFamily(ReplicationSection.COLF);
 
-    return Iterables.concat(replScanner, metaScanner);
+    try {
+      Scanner replScanner = ReplicationTable.getScanner(conn);
+  
+      // Scan only the Status records
+      StatusSection.limit(replScanner);
+
+      // Only look for this specific WAL
+      replScanner.setRange(Range.exact(wal));
+
+      return Iterables.concat(metaScanner, replScanner);
+    } catch (TableNotFoundException e) {
+      // do nothing
+    }
+
+    return metaScanner;
   }
 
   private int scanServers(Map<Path,String> fileToServerMap, Map<String,Path>
nameToFileMap) throws Exception {

http://git-wip-us.apache.org/repos/asf/accumulo/blob/53fc90fc/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
----------------------------------------------------------------------
diff --git a/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
b/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
index 749fc96..1a3995b 100644
--- a/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
+++ b/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
@@ -29,17 +29,19 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Set;
 import java.util.UUID;
 
 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.TableNotFoundException;
 import org.apache.accumulo.core.client.mock.MockInstance;
 import org.apache.accumulo.core.client.security.tokens.PasswordToken;
 import org.apache.accumulo.core.data.Key;
@@ -59,13 +61,13 @@ import org.apache.accumulo.server.replication.ReplicationTable;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.Text;
-import org.easymock.EasyMock;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TestName;
 
+import com.google.common.collect.Iterables;
 import com.google.common.collect.Maps;
 
 public class GarbageCollectWriteAheadLogsTest {
@@ -313,7 +315,7 @@ public class GarbageCollectWriteAheadLogsTest {
     }
 
     @Override
-    protected Iterable<Entry<Key,Value>> getReplicationStatusForFile(Connector
conn, String wal) throws TableNotFoundException {
+    protected Iterable<Entry<Key,Value>> getReplicationStatusForFile(Connector
conn, String wal) {
       return this.replData;
     }
   }
@@ -434,4 +436,64 @@ public class GarbageCollectWriteAheadLogsTest {
     // Both should have been deleted
     Assert.assertEquals(0, nameToFileMap.size());
   }
+
+  @Test
+  public void noReplicationTableDoesntLimitMetatdataResults() throws Exception {
+    Instance inst = new MockInstance(testName.getMethodName());
+    Connector conn = inst.getConnector("root", new PasswordToken(""));
+
+    String wal = "hdfs://localhost:8020/accumulo/wal/tserver+port/123456-1234-1234-12345678";
+    BatchWriter bw = conn.createBatchWriter(MetadataTable.NAME, new BatchWriterConfig());
+    Mutation m = new Mutation(ReplicationSection.getRowPrefix() + wal);
+    m.put(ReplicationSection.COLF, new Text("1"), StatusUtil.newFileValue());
+    bw.addMutation(m);
+    bw.close();
+
+    GarbageCollectWriteAheadLogs gcWALs = new GarbageCollectWriteAheadLogs(inst, volMgr,
false);
+
+    Iterable<Entry<Key,Value>> data = gcWALs.getReplicationStatusForFile(conn,
wal);
+    Entry<Key,Value> entry = Iterables.getOnlyElement(data);
+
+    Assert.assertEquals(ReplicationSection.getRowPrefix() + wal, entry.getKey().getRow().toString());
+  }
+
+  @Test
+  public void fetchesReplicationEntriesFromMetadataAndReplicationTables() throws Exception
{
+    Instance inst = new MockInstance(testName.getMethodName());
+    Connector conn = inst.getConnector("root", new PasswordToken(""));
+    ReplicationTable.create(conn);
+
+    String wal = "hdfs://localhost:8020/accumulo/wal/tserver+port/123456-1234-1234-12345678";
+    BatchWriter bw = conn.createBatchWriter(MetadataTable.NAME, new BatchWriterConfig());
+    Mutation m = new Mutation(ReplicationSection.getRowPrefix() + wal);
+    m.put(ReplicationSection.COLF, new Text("1"), StatusUtil.newFileValue());
+    bw.addMutation(m);
+    bw.close();
+
+    bw = ReplicationTable.getBatchWriter(conn);
+    m = new Mutation(wal);
+    StatusSection.add(m, new Text("1"), StatusUtil.newFileValue());
+    bw.addMutation(m);
+    bw.close();
+
+    GarbageCollectWriteAheadLogs gcWALs = new GarbageCollectWriteAheadLogs(inst, volMgr,
false);
+
+    Iterable<Entry<Key,Value>> iter = gcWALs.getReplicationStatusForFile(conn,
wal);
+    Map<Key,Value> data = new HashMap<>();
+    for (Entry<Key,Value> e : iter) {
+      data.put(e.getKey(), e.getValue());
+    }
+
+    Assert.assertEquals(2, data.size());
+
+    // Should get one element from each table (metadata and replication)
+    for (Key k : data.keySet()) {
+      String row = k.getRow().toString();
+      if (row.startsWith(ReplicationSection.getRowPrefix())) {
+        Assert.assertTrue(row.endsWith(wal));
+      } else {
+        Assert.assertEquals(wal, row);
+      }
+    }
+  }
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/53fc90fc/server/master/src/main/java/org/apache/accumulo/master/replication/ReplicationWorkAssigner.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/replication/ReplicationWorkAssigner.java
b/server/master/src/main/java/org/apache/accumulo/master/replication/ReplicationWorkAssigner.java
index a6607e5..24842a9 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/replication/ReplicationWorkAssigner.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/replication/ReplicationWorkAssigner.java
@@ -177,6 +177,8 @@ public class ReplicationWorkAssigner extends Daemon {
 
       // Keep the state of the work we queued correct
       cleanupFinishedWork();
+
+      UtilWaitThread.sleep(conf.getTimeInMillis(Property.REPLICATION_WORK_ASSIGNMENT_SLEEP));
     }
   }
 

http://git-wip-us.apache.org/repos/asf/accumulo/blob/53fc90fc/test/src/test/java/org/apache/accumulo/test/replication/ReplicationIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/replication/ReplicationIT.java b/test/src/test/java/org/apache/accumulo/test/replication/ReplicationIT.java
index d592716..0d4099c 100644
--- a/test/src/test/java/org/apache/accumulo/test/replication/ReplicationIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/replication/ReplicationIT.java
@@ -108,11 +108,11 @@ public class ReplicationIT extends ConfigurableMacIT {
       
       Scanner s = ReplicationTable.getScanner(connMaster);
       for (Entry<Key,Value> e : s) {
-        System.out.println(e.getKey().toStringNoTruncate() + " " + TextFormat.shortDebugString(Status.parseFrom(e.getValue().get())));
+        log.info(e.getKey().toStringNoTruncate() + " " + TextFormat.shortDebugString(Status.parseFrom(e.getValue().get())));
       }
 
-      System.out.println();
-      System.out.println();
+      log.info("");
+      log.info("");
 
       Thread.sleep(1000);
     }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/53fc90fc/test/src/test/java/org/apache/accumulo/test/replication/ReplicationWithGCIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/replication/ReplicationWithGCIT.java
b/test/src/test/java/org/apache/accumulo/test/replication/ReplicationWithGCIT.java
index 92f6fb3..b20a3bd 100644
--- a/test/src/test/java/org/apache/accumulo/test/replication/ReplicationWithGCIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/replication/ReplicationWithGCIT.java
@@ -55,6 +55,7 @@ import org.apache.accumulo.server.replication.ReplicationTable;
 import org.apache.accumulo.test.functional.ConfigurableMacIT;
 import org.apache.accumulo.tserver.TabletServer;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.RawLocalFileSystem;
 import org.apache.hadoop.io.Text;
@@ -62,6 +63,7 @@ import org.junit.Assert;
 import org.junit.Test;
 
 import com.google.common.collect.Iterables;
+import com.google.protobuf.TextFormat;
 
 /**
  * 
@@ -205,18 +207,6 @@ public class ReplicationWithGCIT extends ConfigurableMacIT {
     }
     bw.close();
 
-    // System.out.println("**** WALs from metadata");
-    // for (String metadataWal : metadataWals) {
-    // System.out.println(metadataWal);
-    // }
-    //
-    // System.out.println("**** WALs from replication");
-    // s = ReplicationTable.getScanner(conn);
-    // StatusSection.limit(s);
-    // for (Entry<Key,Value> entry : s) {
-    // System.out.println(entry.getKey().toStringNoTruncate() + " " + Status.parseFrom(entry.getValue().get()).toString().replace("\n",
", "));
-    // }
-
     s = ReplicationTable.getScanner(conn);
     StatusSection.limit(s);
     bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
@@ -231,13 +221,6 @@ public class ReplicationWithGCIT extends ConfigurableMacIT {
     }
     bw.close();
 
-    // System.out.println("**** WALs from replication");
-    // s = ReplicationTable.getScanner(conn);
-    // StatusSection.limit(s);
-    // for (Entry<Key,Value> entry : s) {
-    // System.out.println(entry.getKey().toStringNoTruncate() + " " + Status.parseFrom(entry.getValue().get()).toString().replace("\n",
", "));
-    // }
-
     // Kill the tserver(s) and restart them
     // to ensure that the WALs we previously observed all move to closed.
     for (ProcessReference proc : cluster.getProcesses().get(ServerType.TABLET_SERVER)) {
@@ -254,24 +237,6 @@ public class ReplicationWithGCIT extends ConfigurableMacIT {
       Entry<Key,Value> entry : s) {}
     }
 
-    // for (int i = 0; i < 5; i++) {
-    // s = conn.createScanner(MetadataTable.NAME, new Authorizations());
-    // s.setRange(ReplicationSection.getRange());
-    // System.out.println("**** Metadata");
-    // for (Entry<Key,Value> entry : s) {
-    // System.out.println(entry.getKey().toStringNoTruncate() + " " + Status.parseFrom(entry.getValue().get()).toString().replace("\n",
", "));
-    // }
-    //
-    // s = ReplicationTable.getScanner(conn);
-    // StatusSection.limit(s);
-    // System.out.println("**** Replication status");
-    // for (Entry<Key,Value> entry : s) {
-    // System.out.println(entry.getKey().toStringNoTruncate() + " " + Status.parseFrom(entry.getValue().get()).toString().replace("\n",
", "));
-    // }
-    //
-    // Thread.sleep(1000);
-    // }
-
     try {
       boolean allClosed = true;
       for (int i = 0; i < 10; i++) {
@@ -306,18 +271,10 @@ public class ReplicationWithGCIT extends ConfigurableMacIT {
       gc.waitFor();
     }
 
-    // System.out.println("****** Replication table iterators");
-    // System.out.println(conn.tableOperations().listIterators(ReplicationTable.NAME));
-    // for (Entry<String,String> entry : conn.tableOperations().getProperties(ReplicationTable.NAME))
{
-    // System.out.println(entry.getKey()+ "=" + entry.getValue());
-    // }
-    // System.out.println();
-
-    System.out.println("****** Final Replication logs before failure");
     s = ReplicationTable.getScanner(conn);
     StatusSection.limit(s);
     for (Entry<Key,Value> entry : s) {
-      System.out.println(entry.getKey().toStringNoTruncate() + " " + Status.parseFrom(entry.getValue().get()).toString().replace("\n",
", "));
+      log.info(entry.getKey().toStringNoTruncate() + " " + TextFormat.shortDebugString(Status.parseFrom(entry.getValue().get())));
     }
     Assert.fail("Expected all replication records to be closed");
   }
@@ -325,6 +282,7 @@ public class ReplicationWithGCIT extends ConfigurableMacIT {
   @Test
   public void replicatedStatusEntriesAreDeleted() throws Exception {
     Connector conn = getConnector();
+    FileSystem fs = FileSystem.getLocal(new Configuration());
     String table1 = "table1";
 
     // replication shouldn't exist when we begin
@@ -469,7 +427,15 @@ public class ReplicationWithGCIT extends ConfigurableMacIT {
       Thread.sleep(1000);
     }
 
-    Assert.assertTrue("Did nto find any replication entries in the replication table", foundResults);
+    Assert.assertTrue("Did not find any replication entries in the replication table", foundResults);
+
+    s = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
+    s.setRange(ReplicationSection.getRange());
+    for (Entry<Key,Value> entry : s) {
+      String row = entry.getKey().getRow().toString();
+      Path file = new Path(row.substring(ReplicationSection.getRowPrefix().length()));
+      Assert.assertTrue(file + " did not exist when it should", fs.exists(file));
+    }
 
     /**
      * After recovery completes, we should have unreplicated, closed Status messages. The
close happens at the beginning of log recovery.

http://git-wip-us.apache.org/repos/asf/accumulo/blob/53fc90fc/test/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/test/src/test/resources/log4j.properties b/test/src/test/resources/log4j.properties
index 71c06e9..dd382f7 100644
--- a/test/src/test/resources/log4j.properties
+++ b/test/src/test/resources/log4j.properties
@@ -34,3 +34,5 @@ log4j.logger.org.apache.accumulo.core.file.rfile.bcfile=INFO
 log4j.logger.org.apache.accumulo.server.util.ReplicationTableUtil=TRACE
 log4j.logger.org.apache.accumulo.core.client.impl.TabletServerBatchReaderIterator=INFO
 log4j.logger.org.apache.accumulo.core.client.impl.ThriftScanner=INFO
+log4j.logger.org.apache.accumulo.server.zookeeper.DistributedWorkQueue=INFO
+log4j.logger.org.apache.accumulo.fate.zookeeper.DistributedReadWriteLock=WARN
\ No newline at end of file


Mime
View raw message