[GitHub] keith-turner closed pull request #845: fixes #823 unreference closed WALs that were never written to

[GitBox <gi...@apache.org>]

keith-turner closed pull request #845: fixes #823 unreference closed WALs that were never written to
URL: https://github.com/apache/accumulo/pull/845
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index bd41541e6a..6acece24b0 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@ -3441,11 +3441,17 @@ public void addNewLogMarker(DfsLogger copy) throws WalMarkerException {
 
   public void walogClosed(DfsLogger currentLog) throws WalMarkerException {
     metadataTableLogs.remove(currentLog);
-    synchronized (closedLogs) {
-      closedLogs.add(currentLog);
+    if (currentLog.getWrites() > 0) {
+      synchronized (closedLogs) {
+        closedLogs.add(currentLog);
+      }
+      log.info("Marking " + currentLog.getPath() + " as closed");
+      walMarker.closeWal(getTabletSession(), currentLog.getPath());
+    } else {
+      log.info(
+          "Marking " + currentLog.getPath() + " as unreferenced (skipping closed writes == 0)");
+      walMarker.walUnreferenced(getTabletSession(), currentLog.getPath());
     }
-    log.info("Marking " + currentLog.getPath() + " as closed");
-    walMarker.closeWal(getTabletSession(), currentLog.getPath());
   }
 
   public void updateBulkImportState(List<String> files, BulkImportState state) {
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/DfsLogger.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/DfsLogger.java
index 263f0c8d26..8f32aaea39 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/DfsLogger.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/DfsLogger.java
@@ -69,6 +69,7 @@
 
 import com.google.common.base.Joiner;
 import com.google.common.base.Optional;
+import com.google.common.base.Preconditions;
 
 /**
  * Wrap a connection to a logger.
@@ -320,6 +321,7 @@ public int hashCode() {
   private AtomicLong syncCounter;
   private AtomicLong flushCounter;
   private final long slowFlushMillis;
+  private long writes = 0;
 
   private DfsLogger(ServerResources conf) {
     this.conf = conf;
@@ -596,6 +598,11 @@ public void close() throws IOException {
       }
   }
 
+  public synchronized long getWrites() {
+    Preconditions.checkState(writes >= 0);
+    return writes;
+  }
+
   public synchronized void defineTablet(long seq, int tid, KeyExtent tablet) throws IOException {
     // write this log to the METADATA table
     final LogFileKey key = new LogFileKey();
@@ -610,12 +617,18 @@ public synchronized void defineTablet(long seq, int tid, KeyExtent tablet) throw
           + " incompatible with this version of Hadoop.");
       throw new RuntimeException(e);
     }
+
+    synchronized (closeLock) {
+      if (closed)
+        throw new LogClosedException();
+    }
   }
 
   private synchronized void write(LogFileKey key, LogFileValue value) throws IOException {
     key.write(encryptingLogFile);
     value.write(encryptingLogFile);
     encryptingLogFile.flush();
+    writes++;
   }
 
   public LoggerOperation log(long seq, int tid, Mutation mutation, Durability durability)
@@ -640,15 +653,16 @@ private LoggerOperation logFileData(List<Pair<LogFileKey,LogFileValue>> keys,
       }
     }
 
-    if (durability == Durability.LOG)
-      return NO_WAIT_LOGGER_OP;
-
     synchronized (closeLock) {
       // use a different lock for close check so that adding to work queue does not need
       // to wait on walog I/O operations
 
       if (closed)
         throw new LogClosedException();
+
+      if (durability == Durability.LOG)
+        return NO_WAIT_LOGGER_OP;
+
       workQueue.add(work);
     }
 
diff --git 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
index 8e946c3eb8..f46c12dfcf 100644
--- 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
@@ -319,15 +319,6 @@ public void run() {
     }));
   }
 
-  public void resetLoggers() throws IOException {
-    logIdLock.writeLock().lock();
-    try {
-      close();
-    } finally {
-      logIdLock.writeLock().unlock();
-    }
-  }
-
   synchronized private void close() throws IOException {
     if (!logIdLock.isWriteLockedByCurrentThread()) {
       throw new IllegalStateException("close should be called with write lock held!");
@@ -453,7 +444,6 @@ boolean test() {
           @Override
           void withWriteLock() throws IOException {
             close();
-            closeForReplication(sessions);
           }
         });
       }
@@ -470,15 +460,10 @@ boolean test() {
       @Override
       void withWriteLock() throws IOException {
         close();
-        closeForReplication(sessions);
       }
     });
   }
 
-  protected void closeForReplication(Collection<CommitSession> sessions) {
-    // TODO We can close the WAL here for replication purposes
-  }
-
   public void defineTablet(final CommitSession commitSession, final Retry writeRetry)
       throws IOException {
     // scribble this into the metadata tablet, too.
@@ -583,5 +568,4 @@ public void recover(VolumeManager fs, KeyExtent extent, TableConfiguration tconf
       throw new IOException(e);
     }
   }
-
 }


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services