geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jinmeil...@apache.org
Subject [16/22] incubator-geode git commit: GEODE-1072: Removing HDFS related code
Date Wed, 27 Apr 2016 21:16:48 GMT
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/46535f28/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSFlushQueueFunction.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSFlushQueueFunction.java b/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSFlushQueueFunction.java
deleted file mode 100644
index cdf7452..0000000
--- a/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSFlushQueueFunction.java
+++ /dev/null
@@ -1,287 +0,0 @@
-/*
- * 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 com.gemstone.gemfire.cache.hdfs.internal.hoplog;
-
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.logging.log4j.Logger;
-
-import com.gemstone.gemfire.cache.asyncqueue.internal.AsyncEventQueueImpl;
-import com.gemstone.gemfire.cache.execute.Function;
-import com.gemstone.gemfire.cache.execute.FunctionContext;
-import com.gemstone.gemfire.cache.execute.FunctionException;
-import com.gemstone.gemfire.cache.execute.FunctionService;
-import com.gemstone.gemfire.cache.execute.RegionFunctionContext;
-import com.gemstone.gemfire.cache.hdfs.internal.FlushObserver.AsyncFlushResult;
-import com.gemstone.gemfire.cache.hdfs.internal.HDFSBucketRegionQueue;
-import com.gemstone.gemfire.distributed.DistributedMember;
-import com.gemstone.gemfire.distributed.internal.InternalDistributedSystem;
-import com.gemstone.gemfire.distributed.internal.ReplyProcessor21;
-import com.gemstone.gemfire.i18n.LogWriterI18n;
-import com.gemstone.gemfire.internal.InternalEntity;
-import com.gemstone.gemfire.internal.cache.ForceReattemptException;
-import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
-import com.gemstone.gemfire.internal.cache.PartitionedRegion;
-import com.gemstone.gemfire.internal.cache.execute.AbstractExecution;
-import com.gemstone.gemfire.internal.cache.execute.LocalResultCollector;
-import com.gemstone.gemfire.internal.cache.wan.AbstractGatewaySender;
-import com.gemstone.gemfire.internal.cache.wan.AbstractGatewaySenderEventProcessor;
-import com.gemstone.gemfire.internal.cache.wan.parallel.ConcurrentParallelGatewaySenderQueue;
-import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
-import com.gemstone.gemfire.internal.logging.LogService;
-import com.gemstone.gemfire.internal.logging.log4j.LocalizedMessage;
-
-public class HDFSFlushQueueFunction implements Function, InternalEntity{
-  private static final int MAX_RETRIES = Integer.getInteger("gemfireXD.maxFlushQueueRetries", 3);
-  private static final boolean VERBOSE = Boolean.getBoolean("hdfsFlushQueueFunction.VERBOSE");
-  private static final Logger logger = LogService.getLogger();
-  private static final String ID = HDFSFlushQueueFunction.class.getName();
-  
-  public static void flushQueue(PartitionedRegion pr, int maxWaitTime) {
-    
-    Set<Integer> buckets = new HashSet<Integer>(pr.getRegionAdvisor().getBucketSet());
-
-    maxWaitTime *= 1000;
-    long start = System.currentTimeMillis();
-    
-    int retries = 0;
-    long remaining = 0;
-    while (retries++ < MAX_RETRIES && (remaining = waitTime(start, maxWaitTime)) > 0) {
-      if (logger.isDebugEnabled() || VERBOSE) {
-        logger.info(LocalizedMessage.create(LocalizedStrings.DEBUG, "Flushing buckets " + buckets 
-            + ", attempt = " + retries 
-            + ", remaining = " + remaining));
-      }
-      
-      HDFSFlushQueueArgs args = new HDFSFlushQueueArgs(buckets, remaining);
-      
-      HDFSFlushQueueResultCollector rc = new HDFSFlushQueueResultCollector(buckets);
-      AbstractExecution exec = (AbstractExecution) FunctionService
-          .onRegion(pr)
-          .withArgs(args)
-          .withCollector(rc);
-      exec.setWaitOnExceptionFlag(true);
-      
-      try {
-        exec.execute(ID);
-        if (rc.getResult()) {
-          if (logger.isDebugEnabled() || VERBOSE) {
-            logger.info(LocalizedMessage.create(LocalizedStrings.DEBUG, "Flushed all buckets successfully")); 
-          }
-          return;
-        }
-      } catch (FunctionException e) {
-        if (logger.isDebugEnabled() || VERBOSE) {
-          logger.info(LocalizedMessage.create(LocalizedStrings.DEBUG, "Encountered error flushing queue"), e); 
-        }
-      }
-      
-      buckets.removeAll(rc.getSuccessfulBuckets());
-      for (int bucketId : buckets) {
-        remaining = waitTime(start, maxWaitTime);
-        if (logger.isDebugEnabled() || VERBOSE) {
-          logger.info(LocalizedMessage.create(LocalizedStrings.DEBUG, "Waiting for bucket " + bucketId)); 
-        }
-        pr.getNodeForBucketWrite(bucketId, new PartitionedRegion.RetryTimeKeeper((int) remaining));
-      }
-    }
-    
-    pr.checkReadiness();
-    throw new FunctionException("Unable to flush the following buckets: " + buckets);
-  }
-  
-  private static long waitTime(long start, long max) {
-    if (max == 0) {
-      return Integer.MAX_VALUE;
-    }
-    return start + max - System.currentTimeMillis();
-  }
-  
-  @Override
-  public void execute(FunctionContext context) {
-    RegionFunctionContext rfc = (RegionFunctionContext) context;
-    PartitionedRegion pr = (PartitionedRegion) rfc.getDataSet();
-    
-    HDFSFlushQueueArgs args = (HDFSFlushQueueArgs) rfc.getArguments();
-    Set<Integer> buckets = new HashSet<Integer>(args.getBuckets());
-    buckets.retainAll(pr.getDataStore().getAllLocalPrimaryBucketIds());
-
-    Map<Integer, AsyncFlushResult> flushes = new HashMap<Integer, AsyncFlushResult>();
-    for (int bucketId : buckets) {
-      try {
-        HDFSBucketRegionQueue brq = getQueue(pr, bucketId);
-        if (brq != null) {
-          if (logger.isDebugEnabled() || VERBOSE) {
-            logger.info(LocalizedMessage.create(LocalizedStrings.DEBUG, "Flushing bucket " + bucketId)); 
-          }
-          flushes.put(bucketId, brq.flush());
-        }
-      } catch (ForceReattemptException e) {
-        if (logger.isDebugEnabled() || VERBOSE) {
-          logger.info(LocalizedMessage.create(LocalizedStrings.DEBUG, "Encountered error flushing bucket " + bucketId), e); 
-        }
-      }
-    }
-    
-    try {
-      long start = System.currentTimeMillis();
-      for (Map.Entry<Integer, AsyncFlushResult> flush : flushes.entrySet()) {
-        long remaining = waitTime(start, args.getMaxWaitTime());
-        if (logger.isDebugEnabled() || VERBOSE) {
-          logger.info(LocalizedMessage.create(LocalizedStrings.DEBUG, "Waiting for bucket " + flush.getKey() 
-              + " to complete flushing, remaining = " + remaining)); 
-        }
-        
-        if (flush.getValue().waitForFlush(remaining, TimeUnit.MILLISECONDS)) {
-          if (logger.isDebugEnabled() || VERBOSE) {
-            logger.info(LocalizedMessage.create(LocalizedStrings.DEBUG, "Bucket " + flush.getKey() + " flushed successfully")); 
-          }
-          rfc.getResultSender().sendResult(new FlushStatus(flush.getKey()));
-        }
-      }
-    } catch (InterruptedException e) {
-      Thread.currentThread().interrupt();
-    }
-    
-    if (logger.isDebugEnabled() || VERBOSE) {
-      logger.info(LocalizedMessage.create(LocalizedStrings.DEBUG, "Sending final flush result")); 
-    }
-    rfc.getResultSender().lastResult(FlushStatus.last());
-  }
-
-  private HDFSBucketRegionQueue getQueue(PartitionedRegion pr, int bucketId) 
-      throws ForceReattemptException {
-    AsyncEventQueueImpl aeq = pr.getHDFSEventQueue();
-    AbstractGatewaySender gw = (AbstractGatewaySender) aeq.getSender();
-    AbstractGatewaySenderEventProcessor ep = gw.getEventProcessor();
-    if (ep == null) {
-      return null;
-    }
-    
-    ConcurrentParallelGatewaySenderQueue queue = (ConcurrentParallelGatewaySenderQueue) ep.getQueue();
-    return queue.getBucketRegionQueue(pr, bucketId);
-  }
-  
-  @Override
-  public String getId() {
-    return ID;
-  }
-
-  @Override
-  public boolean hasResult() {
-    return true;
-  }
-
-  @Override
-  public boolean optimizeForWrite() {
-    return true;
-  }
-
-  @Override
-  public boolean isHA() {
-    return false;
-  }
-  
-  public static class HDFSFlushQueueResultCollector implements LocalResultCollector<Object, Boolean> {
-    private final CountDownLatch complete;
-    private final Set<Integer> expectedBuckets;
-    private final Set<Integer> successfulBuckets;
-
-    private volatile ReplyProcessor21 processor;
-    
-    public HDFSFlushQueueResultCollector(Set<Integer> expectedBuckets) {
-      this.expectedBuckets = expectedBuckets;
-      
-      complete = new CountDownLatch(1);
-      successfulBuckets = new HashSet<Integer>();
-    }
-    
-    public Set<Integer> getSuccessfulBuckets() {
-      synchronized (successfulBuckets) {
-        return new HashSet<Integer>(successfulBuckets);
-      }
-    }
-    
-    @Override
-    public Boolean getResult() throws FunctionException {
-      try {
-        complete.await();
-        synchronized (successfulBuckets) {
-          LogWriterI18n logger = InternalDistributedSystem.getLoggerI18n();
-          if (logger.fineEnabled() || VERBOSE) {
-            logger.info(LocalizedStrings.DEBUG, "Expected buckets: " + expectedBuckets);
-            logger.info(LocalizedStrings.DEBUG, "Successful buckets: " + successfulBuckets);
-          }
-          return expectedBuckets.equals(successfulBuckets);
-        }
-      } catch (InterruptedException e) {
-        Thread.currentThread().interrupt();
-        GemFireCacheImpl.getExisting().getCancelCriterion().checkCancelInProgress(e);
-        throw new FunctionException(e);
-      }
-    }
-
-    @Override
-    public Boolean getResult(long timeout, TimeUnit unit)
-        throws FunctionException, InterruptedException {
-      throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public synchronized void addResult(DistributedMember memberID, Object result) {
-      if (result instanceof FlushStatus) {
-        FlushStatus status = (FlushStatus) result;
-        if (!status.isLast()) {
-          synchronized (successfulBuckets) {
-            successfulBuckets.add(status.getBucketId());
-          }        
-        }
-      }
-    }
-
-    @Override
-    public void endResults() {    	
-      complete.countDown();
-    }
-
-    @Override
-    public void clearResults() {
-    }
-
-    @Override
-    public void setProcessor(ReplyProcessor21 processor) {
-      this.processor = processor;
-    }
-
-    @Override
-    public ReplyProcessor21 getProcessor() {
-      return processor;
-    }
-
-	@Override
-	public void setException(Throwable exception) {
-		// TODO Auto-generated method stub
-		
-	}
-
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/46535f28/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSForceCompactionArgs.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSForceCompactionArgs.java b/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSForceCompactionArgs.java
deleted file mode 100644
index ec0f9ff..0000000
--- a/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSForceCompactionArgs.java
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
- * 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 com.gemstone.gemfire.cache.hdfs.internal.hoplog;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-import java.util.HashSet;
-import java.util.Set;
-
-import com.gemstone.gemfire.DataSerializer;
-import com.gemstone.gemfire.internal.VersionedDataSerializable;
-import com.gemstone.gemfire.internal.Version;
-
-/**
- * Arguments passed to the HDFSForceCompactionFunction
- * 
- */
-@SuppressWarnings("serial")
-public class HDFSForceCompactionArgs implements VersionedDataSerializable {
-
-  private static Version[] serializationVersions = new Version[]{ Version.GFE_81 };
-
-  private HashSet<Integer> buckets;
-
-  private boolean isMajor;
-
-  private int maxWaitTime;
-
-  public HDFSForceCompactionArgs() {
-  }
-
-  public HDFSForceCompactionArgs(Set<Integer> buckets, boolean isMajor, Integer maxWaitTime) {
-    this.buckets = new HashSet<Integer>(buckets);
-    this.isMajor = isMajor;
-    this.maxWaitTime = maxWaitTime;
-  }
-
-  @Override
-  public void toData(DataOutput out) throws IOException {
-    DataSerializer.writeHashSet(buckets, out);
-    out.writeBoolean(isMajor);
-    out.writeInt(maxWaitTime);
-  }
-
-  @Override
-  public void fromData(DataInput in) throws IOException,
-      ClassNotFoundException {
-    this.buckets = DataSerializer.readHashSet(in);
-    this.isMajor = in.readBoolean();
-    this.maxWaitTime = in.readInt();
-  }
-
-  @Override
-  public Version[] getSerializationVersions() {
-    return serializationVersions;
-  }
-
-  public Set<Integer> getBuckets() {
-    return (Set<Integer>) buckets;
-  }
-
-  public void setBuckets(Set<Integer> buckets) {
-    this.buckets = new HashSet<Integer>(buckets);
-  }
-
-  public boolean isMajor() {
-    return isMajor;
-  }
-
-  public void setMajor(boolean isMajor) {
-    this.isMajor = isMajor;
-  }
-
-  public boolean isSynchronous() {
-    return maxWaitTime == 0;
-  }
-
-  public int getMaxWaitTime() {
-    return this.maxWaitTime;
-  }
-
-  @Override
-  public String toString() {
-    StringBuilder sb = new StringBuilder();
-    sb.append(getClass().getCanonicalName()).append("@")
-    .append(System.identityHashCode(this))
-    .append(" buckets:").append(buckets)
-    .append(" isMajor:").append(isMajor)
-    .append(" maxWaitTime:").append(maxWaitTime);
-    return sb.toString();
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/46535f28/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSForceCompactionFunction.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSForceCompactionFunction.java b/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSForceCompactionFunction.java
deleted file mode 100644
index d26ac1b..0000000
--- a/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSForceCompactionFunction.java
+++ /dev/null
@@ -1,129 +0,0 @@
-/*
- * 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 com.gemstone.gemfire.cache.hdfs.internal.hoplog;
-
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-import org.apache.logging.log4j.Logger;
-
-import com.gemstone.gemfire.cache.execute.Function;
-import com.gemstone.gemfire.cache.execute.FunctionContext;
-import com.gemstone.gemfire.cache.execute.RegionFunctionContext;
-import com.gemstone.gemfire.internal.InternalEntity;
-import com.gemstone.gemfire.internal.cache.PartitionedRegion;
-import com.gemstone.gemfire.internal.logging.LogService;
-
-/**
- * Function responsible for forcing a compaction on all members
- * of the system
- *
- */
-@SuppressWarnings("serial")
-public class HDFSForceCompactionFunction implements Function, InternalEntity {
-
-  public static final int FORCE_COMPACTION_MAX_RETRIES = Integer.getInteger("gemfireXD.maxCompactionRetries", 3);
-
-  public static final int BUCKET_ID_FOR_LAST_RESULT = -1;
-
-  public static final String ID = "HDFSForceCompactionFunction";
-
-  private static final Logger logger = LogService.getLogger();
-  
-  @Override
-  public void execute(FunctionContext context) {
-    if (context.isPossibleDuplicate()) {
-      // do not re-execute the function, another function
-      // targeting the failed buckets will be invoked
-      context.getResultSender().lastResult(new CompactionStatus(BUCKET_ID_FOR_LAST_RESULT, false));
-      return;
-    }
-    RegionFunctionContext rfc = (RegionFunctionContext) context;
-    PartitionedRegion pr = (PartitionedRegion) rfc.getDataSet();
-    HDFSForceCompactionArgs args = (HDFSForceCompactionArgs) rfc.getArguments();
-    Set<Integer> buckets = new HashSet<Integer>(args.getBuckets()); // copying avoids race when the function coordinator
-                                                                    // also runs the function locally
-    buckets.retainAll(pr.getDataStore().getAllLocalPrimaryBucketIds());
-
-    List<Future<CompactionStatus>> futures =  pr.forceLocalHDFSCompaction(buckets, args.isMajor(), 0);
-    int waitFor = args.getMaxWaitTime();
-    for (Future<CompactionStatus> future : futures) {
-      long start = System.currentTimeMillis();
-      CompactionStatus status = null;
-      try {
-        // TODO use a CompletionService instead
-        if (!args.isSynchronous() && waitFor <= 0) {
-          break;
-        }
-        status = args.isSynchronous() ? future.get() : future.get(waitFor, TimeUnit.MILLISECONDS);
-        buckets.remove(status.getBucketId());
-        if (logger.isDebugEnabled()) {
-          logger.debug("HDFS: ForceCompaction sending result:"+status);
-        }
-        context.getResultSender().sendResult(status);
-        long elapsedTime = System.currentTimeMillis() - start;
-        waitFor -= elapsedTime;
-      } catch (InterruptedException e) {
-        // send a list of failed buckets after waiting for all buckets
-      } catch (ExecutionException e) {
-        // send a list of failed buckets after waiting for all buckets
-      } catch (TimeoutException e) {
-        // do not wait for other buckets to complete
-        break;
-      }
-    }
-    // for asynchronous invocation, the status is true for buckets that we did not wait for
-    boolean status = args.isSynchronous() ? false : true;
-    for (Integer bucketId : buckets) {
-      if (logger.isDebugEnabled()) {
-        logger.debug("HDFS: ForceCompaction sending result for bucket:"+bucketId);
-      }
-      context.getResultSender().sendResult(new CompactionStatus(bucketId, status));
-    }
-    if (logger.isDebugEnabled()) {
-      logger.debug("HDFS: ForceCompaction sending last result");
-    }
-    context.getResultSender().lastResult(new CompactionStatus(BUCKET_ID_FOR_LAST_RESULT, true));
-  }
-
-  @Override
-  public String getId() {
-    return ID;
-  }
-
-  @Override
-  public boolean hasResult() {
-    return true;
-  }
-
-  @Override
-  public boolean optimizeForWrite() {
-    // run compaction on primary members
-    return true;
-  }
-
-  @Override
-  public boolean isHA() {
-    // so that we can target re-execution on failed buckets
-    return true;
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/46535f28/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSForceCompactionResultCollector.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSForceCompactionResultCollector.java b/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSForceCompactionResultCollector.java
deleted file mode 100644
index ee5e4aa..0000000
--- a/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSForceCompactionResultCollector.java
+++ /dev/null
@@ -1,131 +0,0 @@
-/*
- * 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 com.gemstone.gemfire.cache.hdfs.internal.hoplog;
-
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import com.gemstone.gemfire.cache.execute.FunctionException;
-import com.gemstone.gemfire.distributed.DistributedMember;
-import com.gemstone.gemfire.distributed.internal.ReplyProcessor21;
-import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
-import com.gemstone.gemfire.internal.cache.execute.LocalResultCollector;
-
-/**
- * 
- */
-public class HDFSForceCompactionResultCollector implements LocalResultCollector<Object, List<CompactionStatus>> {
-
-  /** list of received replies*/
-  private List<CompactionStatus> reply = new ArrayList<CompactionStatus>();
-
-  /** semaphore to block the caller of getResult()*/
-  private CountDownLatch waitForResults = new CountDownLatch(1);
-
-  /** boolean to indicate if clearResults() was called to indicate a failure*/
-  private volatile boolean shouldRetry;
-
-  private ReplyProcessor21 processor;
-
-  @Override
-  public List<CompactionStatus> getResult() throws FunctionException {
-    try {
-      waitForResults.await();
-    } catch (InterruptedException e) {
-      Thread.currentThread().interrupt();
-      GemFireCacheImpl.getExisting().getCancelCriterion().checkCancelInProgress(e);
-      throw new FunctionException(e);
-    }
-    return reply;
-  }
-
-  @Override
-  public List<CompactionStatus> getResult(long timeout, TimeUnit unit)
-      throws FunctionException, InterruptedException {
-    throw new UnsupportedOperationException();
-  }
-
-  @Override
-  public void addResult(DistributedMember memberID,
-      Object resultOfSingleExecution) {
-    if (resultOfSingleExecution instanceof CompactionStatus) {
-      CompactionStatus status = (CompactionStatus) resultOfSingleExecution;
-      if (status.getBucketId() != HDFSForceCompactionFunction.BUCKET_ID_FOR_LAST_RESULT) {
-        reply.add(status);
-      }
-    }
-  }
-
-  @Override
-  public void endResults() {
-    waitForResults.countDown();
-  }
-
-  @Override
-  public void clearResults() {
-    this.shouldRetry = true;
-    waitForResults.countDown();
-  }
-
-  /**
-   * @return true if retry should be attempted
-   */
-  public boolean shouldRetry() {
-    return this.shouldRetry || !getFailedBucketIds().isEmpty();
-  }
-
-  private Set<Integer> getFailedBucketIds() {
-    Set<Integer> result = new HashSet<Integer>();
-    for (CompactionStatus status : reply) {
-      if (!status.isStatus()) {
-        result.add(status.getBucketId());
-      }
-    }
-    return result;
-  }
-
-  public Set<Integer> getSuccessfulBucketIds() {
-    Set<Integer> result = new HashSet<Integer>();
-    for (CompactionStatus status : reply) {
-      if (status.isStatus()) {
-        result.add(status.getBucketId());
-      }
-    }
-    return result;
-  }
-
-  @Override
-  public void setProcessor(ReplyProcessor21 processor) {
-    this.processor = processor;
-  }
-
-  @Override
-  public ReplyProcessor21 getProcessor() {
-    return this.processor;
-  }
-
-@Override
-public void setException(Throwable exception) {
-	// TODO Auto-generated method stub
-	
-}
-}
-

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/46535f28/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSLastCompactionTimeFunction.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSLastCompactionTimeFunction.java b/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSLastCompactionTimeFunction.java
deleted file mode 100644
index 789fe4d..0000000
--- a/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSLastCompactionTimeFunction.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * 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 com.gemstone.gemfire.cache.hdfs.internal.hoplog;
-
-import com.gemstone.gemfire.cache.execute.FunctionAdapter;
-import com.gemstone.gemfire.cache.execute.FunctionContext;
-import com.gemstone.gemfire.cache.execute.RegionFunctionContext;
-import com.gemstone.gemfire.internal.InternalEntity;
-import com.gemstone.gemfire.internal.cache.PartitionedRegion;
-
-/**
- * Function that returns the oldest timestamp among all the major
- * compacted buckets on the members
- *
- */
-@SuppressWarnings("serial")
-public class HDFSLastCompactionTimeFunction extends FunctionAdapter implements InternalEntity{
-
-  public static final String ID = "HDFSLastCompactionTimeFunction";
-
-  @Override
-  public void execute(FunctionContext context) {
-    RegionFunctionContext rfc = (RegionFunctionContext) context;
-    PartitionedRegion pr = (PartitionedRegion) rfc.getDataSet();
-    rfc.getResultSender().lastResult(pr.lastLocalMajorHDFSCompaction());
-  }
-
-  @Override
-  public String getId() {
-    return ID;
-  }
-
-  @Override
-  public boolean isHA() {
-    return true;
-  }
-
-  @Override
-  public boolean optimizeForWrite() {
-    return true;
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/46535f28/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSRegionDirector.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSRegionDirector.java b/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSRegionDirector.java
deleted file mode 100644
index 6d70dce..0000000
--- a/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSRegionDirector.java
+++ /dev/null
@@ -1,480 +0,0 @@
-/*
- * 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 com.gemstone.gemfire.cache.hdfs.internal.hoplog;
-
-import java.io.IOException;
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeSet;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-
-import com.gemstone.gemfire.StatisticsFactory;
-import com.gemstone.gemfire.cache.GemFireCache;
-import com.gemstone.gemfire.cache.hdfs.HDFSStore;
-import com.gemstone.gemfire.cache.hdfs.internal.HDFSStoreImpl;
-import com.gemstone.gemfire.cache.hdfs.internal.PersistedEventImpl;
-import com.gemstone.gemfire.i18n.LogWriterI18n;
-import com.gemstone.gemfire.internal.SystemTimer;
-import com.gemstone.gemfire.internal.cache.LocalRegion;
-import com.gemstone.gemfire.internal.cache.persistence.soplog.SortedOplogStatistics;
-import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
-import com.gemstone.gemfire.internal.logging.LogService;
-import com.gemstone.gemfire.internal.logging.log4j.LocalizedMessage;
-
-import org.apache.logging.log4j.Logger;
-
-/**
- * Cache for hoplog organizers associated with buckets of a region. The director creates an
- * instance of organizer on first get request. It does not read HDFS in advance. Creation of
- * organizer depends on File system initialization that takes outside this class. This class also
- * provides utility methods to monitor usage and manage bucket sets.
- * 
- */
-public class HDFSRegionDirector {
-  /*
-   * Maps each region name to its listener and store objects. This map must be populated before file
-   * organizers of a bucket can be created
-   */
-  private final ConcurrentHashMap<String, HdfsRegionManager> regionManagerMap;
-  
-  /**
-   * regions of this Gemfire cache are managed by this director. TODO this
-   * should be final and be provided at the time of creation of this instance or
-   * through a cache directory
-   */
-  private GemFireCache cache;
-  
-  // singleton instance
-  private static HDFSRegionDirector instance;
-  
-  final ScheduledExecutorService janitor;
-  private JanitorTask janitorTask;
-  
-  private static final Logger logger = LogService.getLogger();
-  protected final static String logPrefix = "<" + "RegionDirector" + "> ";
-  
-  
-  private HDFSRegionDirector() {
-    regionManagerMap = new ConcurrentHashMap<String, HDFSRegionDirector.HdfsRegionManager>();
-    janitor = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
-      @Override
-      public Thread newThread(Runnable r) {
-        Thread thread = new Thread(r, "HDFSRegionJanitor");
-        thread.setDaemon(true);
-        return thread;
-      }
-    });
-    
-    long interval = Long.getLong(HoplogConfig.JANITOR_INTERVAL_SECS,
-        HoplogConfig.JANITOR_INTERVAL_SECS_DEFAULT);
-    
-    janitorTask = new JanitorTask();
-    janitor.scheduleWithFixedDelay(janitorTask, interval, interval,
-        TimeUnit.SECONDS);
-  }
-  
-  public synchronized static HDFSRegionDirector getInstance() {
-    if (instance == null) {
-      instance = new HDFSRegionDirector();
-    }
-    return instance;
-  }
-  
-  public HDFSRegionDirector setCache(GemFireCache cache) {
-    this.cache = cache;
-    return this;
-  }
-
-  public GemFireCache getCache() {
-    return this.cache;
-  }
-  /**
-   * Caches listener, store object and list of organizers associated with the region associated with
-   * a region. Subsequently, these objects will be used each time an organizer is created
-   */
-  public synchronized HdfsRegionManager manageRegion(LocalRegion region, String storeName,
-      HoplogListener listener) {
-    
-    HdfsRegionManager manager = regionManagerMap.get(region.getFullPath());
-    if (manager != null) {
-      // this is an attempt to re-register a region. Assuming this was required
-      // to modify listener or hdfs store impl associated with the region. Hence
-      // will clear the region first.
-
-      clear(region.getFullPath());
-    }
-    
-    HDFSStoreImpl store = HDFSStoreDirector.getInstance().getHDFSStore(storeName);
-    manager = new HdfsRegionManager(region, store, listener, getStatsFactory(), this);
-    regionManagerMap.put(region.getFullPath(), manager);
-    
-    if (logger.isDebugEnabled()) {
-      logger.debug("{}Now managing region " + region.getFullPath(), logPrefix);
-    }
-    
-    return manager;
-  }
-  
-  /**
-   * Find the regions that are part of a particular HDFS store.
-   */
-  public Collection<String> getRegionsInStore(HDFSStore store) {
-    TreeSet<String> regions = new TreeSet<String>();
-    for(Map.Entry<String, HdfsRegionManager> entry : regionManagerMap.entrySet()) {
-      if(entry.getValue().getStore().equals(store)) {
-        regions.add(entry.getKey());
-      }
-    }
-    return regions;
-  }
-  
-  public int getBucketCount(String regionPath) {
-    HdfsRegionManager manager = regionManagerMap.get(regionPath);
-    if (manager == null) {
-      throw new IllegalStateException("Region not initialized");
-    }
-
-    return manager.bucketOrganizerMap.size();
-  }
-  
-  public void closeWritersForRegion(String regionPath, int minSizeForFileRollover) throws IOException {
-    regionManagerMap.get(regionPath).closeWriters(minSizeForFileRollover);
-  }
-  /**
-   * removes and closes all {@link HoplogOrganizer} of this region. This call is expected with
-   * a PR disowns a region.
-   */
-  public synchronized void clear(String regionPath) {
-    HdfsRegionManager manager = regionManagerMap.remove(regionPath);
-    if (manager != null) {
-      if (logger.isDebugEnabled()) {
-        logger.debug("{}Closing hoplog region manager for " + regionPath, logPrefix);
-      }
-      manager.close();
-    }
-  }
-
-  /**
-   * Closes all region managers, organizers and hoplogs. This method should be
-   * called before closing the cache to gracefully release all resources
-   */
-  public static synchronized void reset() {
-    if (instance == null) {
-      // nothing to reset
-      return;
-    }
-    
-    instance.janitor.shutdownNow();
-    
-    for (String region : instance.regionManagerMap.keySet()) {
-      instance.clear(region);
-    }
-    instance.cache = null;
-    instance = null;
-  }
-  
-  /**
-   * Terminates current janitor task and schedules a new. The rate of the new
-   * task is based on the value of system property at that time
-   */
-  public static synchronized void resetJanitor() {
-    instance.janitorTask.terminate();
-    instance.janitorTask = instance.new JanitorTask();
-    long interval = Long.getLong(HoplogConfig.JANITOR_INTERVAL_SECS,
-        HoplogConfig.JANITOR_INTERVAL_SECS_DEFAULT);
-    instance.janitor.scheduleWithFixedDelay(instance.janitorTask, 0, interval,
-        TimeUnit.SECONDS);
-  }
-  
-  /**
-   * @param regionPath name of region for which stats object is desired
-   * @return {@link SortedOplogStatistics} instance associated with hdfs region
-   *         name. Null if region is not managed by director
-   */
-  public synchronized SortedOplogStatistics getHdfsRegionStats(String regionPath) {
-    HdfsRegionManager manager = regionManagerMap.get(regionPath);
-    return manager == null ? null : manager.getHdfsStats();
-  }
-  
-  private StatisticsFactory getStatsFactory() {
-    return cache.getDistributedSystem();
-  }
-
-  /**
-   * A helper class to manage region and its organizers
-   */
-  public static class HdfsRegionManager {
-    // name and store configuration of the region whose buckets are managed by this director.
-    private LocalRegion region;
-    private HDFSStoreImpl store;
-    private HoplogListener listener;
-    private volatile boolean closed = false;
-    private final int FILE_ROLLOVER_TASK_INTERVAL = Integer.parseInt
-        (System.getProperty("gemfire.HDFSRegionDirector.FILE_ROLLOVER_TASK_INTERVAL_SECONDS", "60"));
-    
-    private SystemTimer hoplogCloseTimer = null;
-    
-    // instance of hdfs statistics object for this hdfs based region. This
-    // object will collect usage and performance related statistics.
-    private final SortedOplogStatistics hdfsStats;
-
-    /*
-     * An instance of organizer is created for each bucket of regionName region residing on this
-     * node. This member maps bucket id with its corresponding organizer instance. A lock is used to
-     * manage concurrent writes to the map.
-     */
-    private ConcurrentMap<Integer, HoplogOrganizer> bucketOrganizerMap;
-    
-    private HDFSRegionDirector hdfsRegionDirector;
-
-    /**
-     * @param listener
-     *          listener of change events like file creation and deletion
-     * @param hdfsRegionDirector 
-     */
-    HdfsRegionManager(LocalRegion region, HDFSStoreImpl store,
-        HoplogListener listener, StatisticsFactory statsFactory, HDFSRegionDirector hdfsRegionDirector) {
-      bucketOrganizerMap = new ConcurrentHashMap<Integer, HoplogOrganizer>();
-      this.region = region;
-      this.listener = listener;
-      this.store = store;
-      this.hdfsStats = new SortedOplogStatistics(statsFactory, "HDFSRegionStatistics", region.getFullPath());
-      this.hdfsRegionDirector = hdfsRegionDirector;
-    }
-
-    public void closeWriters(int minSizeForFileRollover) throws IOException {
-      final long startTime = System.currentTimeMillis();
-      long elapsedTime = 0;
-        
-      Collection<HoplogOrganizer> organizers = bucketOrganizerMap.values();
-      
-      for (HoplogOrganizer organizer : organizers) {
-      
-        try {
-          this.getRegion().checkReadiness();
-        } catch (Exception e) {
-          break;
-        }
-        
-        ((HDFSUnsortedHoplogOrganizer)organizer).synchronizedCloseWriter(true, 0, 
-            minSizeForFileRollover);
-      }
-      
-    }
-
-    public synchronized <T extends PersistedEventImpl> HoplogOrganizer<T> create(int bucketId) throws IOException {
-      assert !bucketOrganizerMap.containsKey(bucketId);
-
-      HoplogOrganizer<?> organizer = region.getHDFSWriteOnly() 
-          ? new HDFSUnsortedHoplogOrganizer(this, bucketId) 
-          : new HdfsSortedOplogOrganizer(this, bucketId);
-
-      bucketOrganizerMap.put(bucketId, organizer);
-      // initialize a timer that periodically closes the hoplog writer if the 
-      // time for rollover has passed. It also has the responsibility to fix the files.  
-      if (this.region.getHDFSWriteOnly() && 
-          hoplogCloseTimer == null) {
-        hoplogCloseTimer = new SystemTimer(hdfsRegionDirector.
-            getCache().getDistributedSystem(), true);
-        
-        // schedule the task to fix the files that were not closed properly 
-        // last time. 
-        hoplogCloseTimer.scheduleAtFixedRate(new CloseTmpHoplogsTimerTask(this), 
-            1000, FILE_ROLLOVER_TASK_INTERVAL * 1000);
-        
-        if (logger.isDebugEnabled()) {
-          logger.debug("{}Schedulng hoplog rollover timer with interval "+ FILE_ROLLOVER_TASK_INTERVAL + 
-              " for hoplog organizer for " + region.getFullPath()
-              + ":" + bucketId + " " + organizer, logPrefix);
-        }
-      }
-      
-      if (logger.isDebugEnabled()) {
-        logger.debug("{}Constructed hoplog organizer for " + region.getFullPath()
-            + ":" + bucketId + " " + organizer, logPrefix);
-      }
-      return (HoplogOrganizer<T>) organizer;
-    }
-    
-    public synchronized <T extends PersistedEventImpl> void addOrganizer(
-        int bucketId, HoplogOrganizer<T> organizer) {
-      if (bucketOrganizerMap.containsKey(bucketId)) {
-        throw new IllegalArgumentException();
-      }
-      if (logger.isDebugEnabled()) {
-        logger.debug("{}added pre constructed organizer " + region.getFullPath()
-            + ":" + bucketId + " " + organizer, logPrefix);
-      }
-      bucketOrganizerMap.put(bucketId, organizer);
-    }
-
-    public void close() {
-      closed = true;
-      
-      if (this.region.getHDFSWriteOnly() && 
-          hoplogCloseTimer != null) {
-        hoplogCloseTimer.cancel();
-        hoplogCloseTimer = null;
-      }
-      for (int bucket : bucketOrganizerMap.keySet()) {
-        close(bucket);
-      }
-    }
-    
-    public boolean isClosed() {
-      return closed;
-    }
-
-    public synchronized void close(int bucketId) {
-      try {
-        HoplogOrganizer organizer = bucketOrganizerMap.remove(bucketId);
-        if (organizer != null) {
-          if (logger.isDebugEnabled()) {
-            logger.debug("{}Closing hoplog organizer for " + region.getFullPath() + ":" + 
-                bucketId + " " + organizer, logPrefix);
-          }
-          organizer.close();
-        }
-      } catch (IOException e) {
-        if (logger.isDebugEnabled()) {
-          logger.debug(logPrefix + "Error closing hoplog organizer for " + region.getFullPath() + ":" + bucketId, e);
-        }
-      }
-      //TODO abort compaction and flush requests for this region
-    }
-    
-    public static String getRegionFolder(String regionPath) {
-      String folder = regionPath;
-      //Change any underscore into a double underscore
-      folder = folder.replace("_", "__");
-      //get rid of the leading slash
-      folder = folder.replaceFirst("^/", "");
-      //replace slashes with underscores
-      folder = folder.replace('/', '_');
-      return folder;
-    }
-
-    public String getRegionFolder() {
-      return getRegionFolder(region.getFullPath());
-    }
-
-    public HoplogListener getListener() {
-      return listener;
-    }
-
-    public HDFSStoreImpl getStore() {
-      return store;
-    }
-
-    public LocalRegion getRegion() {
-      return region;
-    }
-    
-    public SortedOplogStatistics getHdfsStats() {
-      return hdfsStats;
-    }
-    
-    public Collection<HoplogOrganizer> getBucketOrganizers(){
-      return this.bucketOrganizerMap.values();
-    }
-
-    /**
-     * get the HoplogOrganizers only for the given set of buckets
-     */
-    public Collection<HoplogOrganizer> getBucketOrganizers(Set<Integer> buckets){
-      Set<HoplogOrganizer> result = new HashSet<HoplogOrganizer>();
-      for (Integer bucketId : buckets) {
-        result.add(this.bucketOrganizerMap.get(bucketId));
-      }
-      return result;
-    }
-
-    /**
-     * Delete all files from HDFS for this region. This method
-     * should be called after all members have destroyed their
-     * region in gemfire, so there should be no threads accessing
-     * these files.
-     * @throws IOException 
-     */
-    public void destroyData() throws IOException {
-      //Make sure everything is shut down and closed.
-      close();
-      if (store == null) {
-        return;
-      }
-      Path regionPath = new Path(store.getHomeDir(), getRegionFolder());
-      
-      //Delete all files in HDFS.
-      FileSystem fs = getStore().getFileSystem();
-      if(!fs.delete(regionPath, true)) {
-        if(fs.exists(regionPath)) {
-          throw new IOException("Unable to delete " + regionPath);
-        }
-      }
-    }
-
-    public void performMaintenance() throws IOException {
-      Collection<HoplogOrganizer> buckets = getBucketOrganizers();
-      for (HoplogOrganizer bucket : buckets) {
-        bucket.performMaintenance();
-      }
-    }
-  }
-  
-  private class JanitorTask implements Runnable {
-    boolean terminated = false;
-    @Override
-    public void run() {
-      if (terminated) {
-        return;
-      }
-      fineLog("Executing HDFS Region janitor task", null);
-      
-      Collection<HdfsRegionManager> regions = regionManagerMap.values();
-      for (HdfsRegionManager region : regions) {
-        fineLog("Maintaining region:" + region.getRegionFolder(), null);
-        try {
-          region.performMaintenance();
-        } catch (Throwable e) {
-          logger.info(LocalizedMessage.create(LocalizedStrings.HOPLOG_IO_ERROR , region.getRegionFolder()));
-          logger.info(LocalizedMessage.create(LocalizedStrings.ONE_ARG, e.getMessage()));
-          fineLog(null, e);
-        }
-      }
-    }
-
-    public void terminate() {
-      terminated = true;
-    }
-  }
-  
-  protected static void fineLog(String message, Throwable e) {
-    if(logger.isDebugEnabled()) {
-      logger.debug(message, e);
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/46535f28/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSStoreDirector.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSStoreDirector.java b/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSStoreDirector.java
deleted file mode 100644
index 880ef3e..0000000
--- a/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSStoreDirector.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * 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 com.gemstone.gemfire.cache.hdfs.internal.hoplog;
-
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.concurrent.ConcurrentHashMap;
-
-
-import com.gemstone.gemfire.cache.hdfs.internal.HDFSStoreImpl;
-
-/**
- * HDFSStoreDirector is created for managing all instances of HDFSStoreImpl.    
- *
- */
-public final class HDFSStoreDirector {
-  private final ConcurrentHashMap<String, HDFSStoreImpl> storeMap = new ConcurrentHashMap<String, HDFSStoreImpl>();
-
-  // singleton instance
-  private static volatile HDFSStoreDirector instance;
-  
-  private HDFSStoreDirector() {
-
-  }
-  
-  public static final HDFSStoreDirector getInstance() {
-    if (instance == null) {
-      synchronized (HDFSStoreDirector.class)  {
-        if (instance == null)
-          instance = new HDFSStoreDirector();
-      }
-    }
-    return instance;
-  }
-
-  // Called when the region is created.
-  public final void addHDFSStore(HDFSStoreImpl hdfsStore){
-    this.storeMap.put(hdfsStore.getName(), hdfsStore); 
-  }
-  
-  public final HDFSStoreImpl getHDFSStore(String hdfsStoreName) {
-    return this.storeMap.get(hdfsStoreName);
-  }
-  
-  public final void removeHDFSStore(String hdfsStoreName) {
-    this.storeMap.remove(hdfsStoreName);
-  } 
-  
-  public void closeHDFSStores() {
-    Iterator<HDFSStoreImpl> it = this.storeMap.values().iterator();
-    while (it.hasNext()) {
-      HDFSStoreImpl hsi = it.next();
-      hsi.close();
-    }
-    this.storeMap.clear();
-  }
-
-   public ArrayList<HDFSStoreImpl> getAllHDFSStores() {
-    ArrayList<HDFSStoreImpl> hdfsStores = new ArrayList<HDFSStoreImpl>();
-    hdfsStores.addAll(this.storeMap.values());
-    return hdfsStores;
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/46535f28/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSUnsortedHoplogOrganizer.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSUnsortedHoplogOrganizer.java b/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSUnsortedHoplogOrganizer.java
deleted file mode 100644
index cbb35cb..0000000
--- a/geode-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/hoplog/HDFSUnsortedHoplogOrganizer.java
+++ /dev/null
@@ -1,447 +0,0 @@
-/*
- * 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 com.gemstone.gemfire.cache.hdfs.internal.hoplog;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.concurrent.Callable;
-import java.util.concurrent.Future;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.PathFilter;
-
-import com.gemstone.gemfire.cache.CacheClosedException;
-import com.gemstone.gemfire.cache.hdfs.internal.QueuedPersistentEvent;
-import com.gemstone.gemfire.cache.hdfs.internal.UnsortedHoplogPersistedEvent;
-import com.gemstone.gemfire.cache.hdfs.internal.hoplog.HDFSRegionDirector.HdfsRegionManager;
-import com.gemstone.gemfire.cache.hdfs.internal.hoplog.Hoplog.HoplogWriter;
-import com.gemstone.gemfire.internal.HeapDataOutputStream;
-import com.gemstone.gemfire.internal.cache.ForceReattemptException;
-import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
-import com.gemstone.gemfire.internal.logging.log4j.LocalizedMessage;
-import org.apache.hadoop.hbase.util.FSUtils;
-
-/**
- * Manages unsorted Hoplog files for a bucket (Streaming Ingest option). An instance per bucket 
- * will exist in each PR
- * 
- *
- */
-public class HDFSUnsortedHoplogOrganizer extends AbstractHoplogOrganizer<UnsortedHoplogPersistedEvent> {
-  public static final String HOPLOG_REGEX = HOPLOG_NAME_REGEX + "("
-      + SEQ_HOPLOG_EXTENSION + "|" + TEMP_HOPLOG_EXTENSION + ")";
-  public static final Pattern HOPLOG_PATTERN = Pattern.compile(HOPLOG_REGEX);
-  protected static String TMP_FILE_NAME_REGEX = HOPLOG_NAME_REGEX + SEQ_HOPLOG_EXTENSION + TEMP_HOPLOG_EXTENSION + "$";
-  protected static final Pattern patternForTmpHoplog = Pattern.compile(TMP_FILE_NAME_REGEX);
-  
-   volatile private HoplogWriter writer;
-   volatile private Hoplog currentHoplog;
-   
-   volatile private long lastFlushTime = System.currentTimeMillis();
-   
-   volatile private boolean abortFlush = false;
-   private FileSystem fileSystem;
-   
-   public HDFSUnsortedHoplogOrganizer(HdfsRegionManager region, int bucketId) throws IOException{
-    super(region, bucketId);
-    writer = null;
-    sequence = new AtomicInteger(0);
-
-    fileSystem = store.getFileSystem();
-    if (! fileSystem.exists(bucketPath)) {
-      return;
-    }
-    
-    FileStatus validHoplogs[] = FSUtils.listStatus(fileSystem, bucketPath, new PathFilter() {
-      @Override
-      public boolean accept(Path file) {
-        // All valid hoplog files must match the regex
-        Matcher matcher = HOPLOG_PATTERN.matcher(file.getName());
-        return matcher.matches();
-      }
-    });
-
-    if (validHoplogs != null && validHoplogs.length > 0) {
-      for (FileStatus file : validHoplogs) {
-        // account for the disk used by this file
-        incrementDiskUsage(file.getLen());
-      }
-    }
-
-  }
-  
-    @Override
-    public void close() throws IOException {
-      super.close();
-      if (logger.isDebugEnabled())
-        logger.debug("{}Closing the hoplog organizer and the open files", logPrefix);
-      // abort the flush so that we can immediately call the close current writer. 
-      abortFlush = true;
-      synchronizedCloseWriter(true, 0, 0);
-    }
-    
-    
-    /**
-     * Flushes the data to HDFS. 
-     * Synchronization ensures that the writer is not closed when flush is happening.
-     * To abort the flush, abortFlush needs to be set.  
-     * @throws ForceReattemptException 
-     */
-     @Override
-    public synchronized void flush(Iterator<? extends QueuedPersistentEvent> bufferIter, final int count)
-        throws IOException, ForceReattemptException {
-      assert bufferIter != null;
-      
-      if (abortFlush)
-        throw new CacheClosedException("Either the region has been cleared " +
-            "or closed. Aborting the ongoing flush operation.");
-      if (logger.isDebugEnabled())
-        logger.debug("{}Initializing flush operation", logPrefix);
-      
-      // variables for updating stats
-      long start = stats.getFlush().begin();
-      int byteCount = 0;
-      if (writer == null) {
-        // Hoplogs of sequence files are always created with a 0 sequence number
-        currentHoplog = getTmpSortedOplog(0, SEQ_HOPLOG_EXTENSION);
-        try {
-          writer = this.store.getSingletonWriter().runSerially(new Callable<Hoplog.HoplogWriter>() {
-            @Override
-            public HoplogWriter call() throws Exception {
-              return currentHoplog.createWriter(count);
-            }
-          });
-        } catch (Exception e) {
-          if (e instanceof IOException) {
-            throw (IOException)e;
-          }
-          throw new IOException(e);
-        }
-      }
-      long timeSinceLastFlush = (System.currentTimeMillis() - lastFlushTime)/1000 ;
-      
-      try {
-        /**MergeGemXDHDFSToGFE changed the following statement as the code of HeapDataOutputStream is not merged */
-        //HeapDataOutputStream out = new HeapDataOutputStream();
-        while (bufferIter.hasNext()) {
-          HeapDataOutputStream out = new HeapDataOutputStream(1024, null);
-          if (abortFlush) {
-            stats.getFlush().end(byteCount, start);
-            throw new CacheClosedException("Either the region has been cleared " +
-            		"or closed. Aborting the ongoing flush operation.");
-          }
-          QueuedPersistentEvent item = bufferIter.next();
-          item.toHoplogEventBytes(out);
-          byte[] valueBytes = out.toByteArray();
-          writer.append(item.getRawKey(), valueBytes);
-          // add key length and value length to stats byte counter
-          byteCount += (item.getRawKey().length + valueBytes.length);
-          /**MergeGemXDHDFSToGFE how to clear for reuse. Leaving it for Darrel to merge this change*/
-          //out.clearForReuse();
-        }
-        // ping secondaries before making the file a legitimate file to ensure 
-        // that in case of split brain, no other vm has taken up as primary. #50110. 
-        if (!abortFlush)
-          pingSecondaries();
-        // append completed. If the file is to be rolled over, 
-        // close writer and rename the file to a legitimate name.
-        // Else, sync the already written data with HDFS nodes. 
-        int maxFileSize = this.store.getWriteOnlyFileRolloverSize() * 1024 * 1024;  
-        int fileRolloverInterval = this.store.getWriteOnlyFileRolloverInterval(); 
-        if (writer.getCurrentSize() >= maxFileSize || 
-            timeSinceLastFlush >= fileRolloverInterval) {
-          closeCurrentWriter();
-        }
-        else {
-          // if flush is not aborted, hsync the batch. It ensures that 
-          // the batch has reached HDFS and we can discard it. 
-          if (!abortFlush)
-            writer.hsync();
-        }
-      } catch (IOException e) {
-        stats.getFlush().error(start);
-        // as there is an exception, it can be probably be a file specific problem.
-        // close the current file to avoid any file specific issues next time  
-        closeCurrentWriter();
-        // throw the exception so that async queue will dispatch the same batch again 
-        throw e;
-      } 
-      
-      stats.getFlush().end(byteCount, start);
-    }
-    
-    /**
-     * Synchronization ensures that the writer is not closed when flush is happening. 
-     */
-    synchronized void synchronizedCloseWriter(boolean forceClose, 
-        long timeSinceLastFlush, int minsizeforrollover) throws IOException { 
-      long writerSize = 0;
-      if (writer != null){
-        writerSize = writer.getCurrentSize();
-      }
-      
-      if (writerSize < (minsizeforrollover * 1024L))
-        return;
-      
-      int maxFileSize = this.store.getWriteOnlyFileRolloverSize() * 1024 * 1024;  
-      int fileRolloverInterval = this.store.getWriteOnlyFileRolloverInterval(); 
-      if (writerSize >= maxFileSize || 
-          timeSinceLastFlush >= fileRolloverInterval || forceClose) {
-        closeCurrentWriter();
-      }
-      }
-        
-    
-    /**
-     * Closes the current writer so that next time a new hoplog can 
-     * be created. Also, fixes any tmp hoplogs. 
-     * 
-     * @throws IOException
-     */
-    void closeCurrentWriter() throws IOException {
-      
-      if (writer != null) {
-        // If this organizer is closing, it is ok to ignore exceptions here
-        // because CloseTmpHoplogsTimerTask
-        // on another member may have already renamed the hoplog
-        // fixes bug 49141
-        boolean isClosing = abortFlush;
-        try {
-          incrementDiskUsage(writer.getCurrentSize());
-        } catch (IOException e) {
-          if (!isClosing) {
-            throw e;
-          }
-        }
-        if (logger.isDebugEnabled())
-          logger.debug("{}Closing hoplog " + currentHoplog.getFileName(), logPrefix);
-        try{
-          writer.close();
-          makeLegitimate(currentHoplog);
-        } catch (IOException e) {
-          if (!isClosing) {
-            logger.warn(LocalizedStrings.HOPLOG_FLUSH_OPERATION_FAILED, e);
-            throw e;
-          }
-        } finally {
-          writer = null;
-          lastFlushTime = System.currentTimeMillis();
-        }
-      }
-      else
-        lastFlushTime = System.currentTimeMillis();
-    }
-
-    @Override
-    public void clear() throws IOException {
-      boolean prevAbortFlushFlag = abortFlush;
-      // abort the flush so that we can immediately call the close current writer. 
-      abortFlush = true;
-      
-      // Close if there is any existing writer. 
-      try {
-        synchronizedCloseWriter(true, 0, 0);
-      } catch (IOException e) {
-        logger.warn(LocalizedStrings.HOPLOG_CLOSE_FAILED, e);
-      }
-      
-      // reenable the aborted flush
-      abortFlush = prevAbortFlushFlag;
-      
-      // Mark the hoplogs for deletion
-      markHoplogsForDeletion();
-      
-    }
-  
-    @Override
-    public void performMaintenance() {
-      // TODO remove the timer for tmp file conversion. Use this instead
-    }
-
-    @Override
-    public Future<CompactionStatus> forceCompaction(boolean isMajor) {
-      return null;
-    }
-
-    @Override
-    protected Hoplog getHoplog(Path hoplogPath) throws IOException {
-      Hoplog so = new SequenceFileHoplog(fileSystem, hoplogPath, stats);
-      return so;
-    }
-  
-  /**
-   * Fixes the size of hoplogs that were not closed properly last time. 
-   * Such hoplogs are *.tmphop files. Identify them and open them and close 
-   * them, this fixes the size. After doing this rename them to *.hop. 
-   * 
-   * @throws IOException
-   * @throws ForceReattemptException 
-   */
-  void identifyAndFixTmpHoplogs(FileSystem fs) throws IOException, ForceReattemptException {
-    if (logger.isDebugEnabled())
-      logger.debug("{}Fixing temporary hoplogs", logPrefix);
-    
-    // A different filesystem is passed to this function for the following reason: 
-    // For HDFS, if a file wasn't closed properly last time, 
-    // while calling FileSystem.append for this file, FSNamesystem.startFileInternal->
-    // FSNamesystem.recoverLeaseInternal function gets called. 
-    // This function throws AlreadyBeingCreatedException if there is an open handle, to any other file, 
-    // created using the same FileSystem object. This is a bug and is being tracked at: 
-    // https://issues.apache.org/jira/browse/HDFS-3848?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
-    // 
-    // The fix for this bug is not yet part of Pivotal HD. So to overcome the bug, 
-    // we create a new file system for the timer task so that it does not encounter the bug. 
-    
-    FileStatus tmpHoplogs[] = FSUtils.listStatus(fs, fs.makeQualified(bucketPath), new PathFilter() {
-      @Override
-      public boolean accept(Path file) {
-        // All valid hoplog files must match the regex
-        Matcher matcher = patternForTmpHoplog.matcher(file.getName());
-        return matcher.matches();
-      }
-    });
-    
-    if (tmpHoplogs == null || tmpHoplogs.length == 0) {
-      if (logger.isDebugEnabled())
-        logger.debug("{}No files to fix", logPrefix);
-      return;
-    }
-    // ping secondaries so that in case of split brain, no other vm has taken up 
-    // as primary. #50110. 
-    pingSecondaries();
-    if (logger.isDebugEnabled())
-      logger.debug("{}Files to fix " + tmpHoplogs.length, logPrefix);
-
-    String currentHoplogName = null;
-    // get the current hoplog name. We need to ignore current hoplog while fixing. 
-    if (currentHoplog != null) {
-      currentHoplogName = currentHoplog.getFileName();
-    }
-    
-    for (int i = 0; i < tmpHoplogs.length; i++) {
-      // Skip directories
-      if (tmpHoplogs[i].isDirectory()) {
-        continue;
-      }
-
-      final Path p = tmpHoplogs[i].getPath();
-      
-      if (tmpHoplogs[i].getPath().getName().equals(currentHoplogName)){
-        if (logger.isDebugEnabled())
-          logger.debug("Skipping current file: " + tmpHoplogs[i].getPath().getName(), logPrefix);
-        continue;
-      } 
-      
-      SequenceFileHoplog hoplog = new SequenceFileHoplog(fs, p, stats);
-      try {
-        makeLegitimate(hoplog);
-        logger.info (LocalizedMessage.create(LocalizedStrings.DEBUG, "Hoplog " + p + " was a temporary " +
-            "hoplog because the node managing it wasn't shutdown properly last time. Fixed the hoplog name."));
-      } catch (IOException e) {
-        logger.info (LocalizedMessage.create(LocalizedStrings.DEBUG, "Hoplog " + p + " is still a temporary " +
-            "hoplog because the node managing it wasn't shutdown properly last time. Failed to " +
-            "change the hoplog name because an exception was thrown while fixing it. " + e));
-      }
-    }
-  }
-  
-  private FileStatus[] getExpiredHoplogs() throws IOException {
-    FileStatus files[] = FSUtils.listStatus(fileSystem, bucketPath, new PathFilter() {
-      @Override
-      public boolean accept(Path file) {
-        // All expired hoplog end with expire extension and must match the valid file regex
-        String fileName = file.getName();
-        if (! fileName.endsWith(EXPIRED_HOPLOG_EXTENSION)) {
-          return false;
-        }
-        return true;
-      }
-    });
-    return files;
-  }
-  /**
-   * locks sorted oplogs collection, removes oplog and renames for deletion later
-   * @throws IOException 
-   */
-  private void markHoplogsForDeletion() throws IOException {
-    
-    ArrayList<IOException> errors = new ArrayList<IOException>();
-    FileStatus validHoplogs[] = FSUtils.listStatus(fileSystem, bucketPath, new PathFilter() {
-      @Override
-      public boolean accept(Path file) {
-        // All valid hoplog files must match the regex
-        Matcher matcher = HOPLOG_PATTERN.matcher(file.getName());
-        return matcher.matches();
-      }
-    });
-    
-    FileStatus[] expired = getExpiredHoplogs();
-    validHoplogs = filterValidHoplogs(validHoplogs, expired);
-
-    if (validHoplogs == null || validHoplogs.length == 0) {
-      return;
-    }
-    for (FileStatus fileStatus : validHoplogs) {
-      try {
-        addExpiryMarkerForAFile(getHoplog(fileStatus.getPath()));
-      } catch (IOException e) {
-        // even if there is an IO error continue removing other hoplogs and
-        // notify at the end
-        errors.add(e);
-      }
-    }
-    
-    if (!errors.isEmpty()) {
-      for (IOException e : errors) {
-        logger.warn(LocalizedStrings.HOPLOG_HOPLOG_REMOVE_FAILED, e);
-      }
-    }
-  }
-  
-  @Override
-  public Compactor getCompactor() {
-    throw new UnsupportedOperationException("Not supported for " + this.getClass().getSimpleName());
-  }
-  
-    @Override
-  public HoplogIterator<byte[], UnsortedHoplogPersistedEvent> scan(
-      long startOffset, long length) throws IOException {
-    throw new UnsupportedOperationException("Not supported for " + this.getClass().getSimpleName());
-    }
-
-  public long getLastFlushTime() {
-    return this.lastFlushTime;
-      }
-  
-  public long getfileRolloverInterval(){
-    int fileRolloverInterval = this.store.getWriteOnlyFileRolloverInterval(); 
-    return fileRolloverInterval;
-    }
-
-  @Override
-  public long getLastMajorCompactionTimestamp() {
-    throw new UnsupportedOperationException();
-  }
-
-}


Mime
View raw message