hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jdcry...@apache.org
Subject svn commit: r1501503 - in /hbase/trunk: hbase-client/src/main/java/org/apache/hadoop/hbase/client/replication/ hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/ hba...
Date Tue, 09 Jul 2013 19:59:39 GMT
Author: jdcryans
Date: Tue Jul  9 19:59:39 2013
New Revision: 1501503

URL: http://svn.apache.org/r1501503
Log:
HBASE-8861  Remove ReplicationState completely (Chris Trezzo via JD)

Removed:
    hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationStateImpl.java
    hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationStateInterface.java
    hbase/trunk/hbase-server/src/main/ruby/shell/commands/start_replication.rb
    hbase/trunk/hbase-server/src/main/ruby/shell/commands/stop_replication.rb
Modified:
    hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationAdmin.java
    hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationStateZKBase.java
    hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationZookeeper.java
    hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplication.java
    hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/master/ReplicationLogCleaner.java
    hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/Replication.java
    hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java
    hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
    hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java
    hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java
    hbase/trunk/hbase-server/src/main/javadoc/org/apache/hadoop/hbase/replication/package.html
    hbase/trunk/hbase-server/src/main/ruby/hbase/replication_admin.rb
    hbase/trunk/hbase-server/src/main/ruby/shell.rb
    hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestLogsCleaner.java
    hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java
    hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationBase.java
    hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationSmallTests.java
    hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationStateZKImpl.java
    hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationZookeeper.java
    hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceManager.java

Modified: hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationAdmin.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationAdmin.java?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationAdmin.java
(original)
+++ hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationAdmin.java
Tue Jul  9 19:59:39 2013
@@ -52,16 +52,6 @@ import java.util.Map;
  * used to keep track of the replication state.
  * </p>
  * <p>
- * Enabling and disabling peers is currently not supported.
- * </p>
- * <p>
- * As cluster replication is still experimental, a kill switch is provided
- * in order to stop all replication-related operations, see
- * {@link #setReplicating(boolean)}. When setting it back to true, the new
- * state of all the replication streams will be unknown and may have holes.
- * Use at your own risk.
- * </p>
- * <p>
  * To see which commands are available in the shell, type
  * <code>replication</code>.
  * </p>
@@ -163,36 +153,6 @@ public class ReplicationAdmin implements
   }
 
   /**
-   * Get the current status of the kill switch, if the cluster is replicating
-   * or not.
-   * @return true if the cluster is replicated, otherwise false
-   */
-  public boolean getReplicating() throws IOException {
-    try {
-      return this.replicationZk.getReplication();
-    } catch (KeeperException e) {
-      throw new IOException("Couldn't get the replication status");
-    }
-  }
-
-  /**
-   * Kill switch for all replication-related features
-   * @param newState true to start replication, false to stop it.
-   * completely
-   * @return the previous state
-   */
-  public boolean setReplicating(boolean newState) throws IOException {
-    boolean prev = true;
-    try {
-      prev = getReplicating();
-      this.replicationZk.setReplication(newState);
-    } catch (KeeperException e) {
-      throw new IOException("Unable to set the replication state", e);
-    }
-    return prev;
-  }
-
-  /**
    * Get the ZK-support tool created and used by this object for replication.
    * @return the ZK-support tool
    */

Modified: hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationStateZKBase.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationStateZKBase.java?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationStateZKBase.java
(original)
+++ hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationStateZKBase.java
Tue Jul  9 19:59:39 2013
@@ -39,8 +39,6 @@ public abstract class ReplicationStateZK
    * cluster.
    */
   protected final String peerStateNodeName;
-  /** The name of the znode that contains the replication status of the local cluster. */
-  protected final String stateZNode;
   /** The name of the base znode that contains all replication state. */
   protected final String replicationZNode;
   /** The name of the znode that contains a list of all remote slave (i.e. peer) clusters.
*/
@@ -68,11 +66,9 @@ public abstract class ReplicationStateZK
     String replicationZNodeName = conf.get("zookeeper.znode.replication", "replication");
     String peersZNodeName = conf.get("zookeeper.znode.replication.peers", "peers");
     String queuesZNodeName = conf.get("zookeeper.znode.replication.rs", "rs");
-    String stateZNodeName = conf.get("zookeeper.znode.replication.state", "state");
     this.peerStateNodeName = conf.get("zookeeper.znode.replication.peers.state", "peer-state");
     this.ourClusterKey = ZKUtil.getZooKeeperClusterKey(this.conf);
     this.replicationZNode = ZKUtil.joinZNode(this.zookeeper.baseZNode, replicationZNodeName);
-    this.stateZNode = ZKUtil.joinZNode(replicationZNode, stateZNodeName);
     this.peersZNode = ZKUtil.joinZNode(replicationZNode, peersZNodeName);
     this.queuesZNode = ZKUtil.joinZNode(replicationZNode, queuesZNodeName);
   }
@@ -90,9 +86,8 @@ public abstract class ReplicationStateZK
   /**
    * @param state
    * @return Serialized protobuf of <code>state</code> with pb magic prefix prepended
suitable for
-   *         use as content of either the cluster state znode -- whether or not we should
be
-   *         replicating kept in /hbase/replication/state -- or as content of a peer-state
znode
-   *         under a peer cluster id as in /hbase/replication/peers/PEER_ID/peer-state.
+   *         use as content of a peer-state znode under a peer cluster id as in
+   *         /hbase/replication/peers/PEER_ID/peer-state.
    */
   protected static byte[] toByteArray(final ZooKeeperProtos.ReplicationState.State state)
{
     byte[] bytes =

Modified: hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationZookeeper.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationZookeeper.java?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationZookeeper.java
(original)
+++ hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationZookeeper.java
Tue Jul  9 19:59:39 2013
@@ -70,7 +70,7 @@ import java.util.concurrent.atomic.Atomi
  * </pre>
  */
 @InterfaceAudience.Private
-public class ReplicationZookeeper extends ReplicationStateZKBase implements Closeable {
+public class ReplicationZookeeper extends ReplicationStateZKBase {
   private static final Log LOG = LogFactory.getLog(ReplicationZookeeper.class);
 
   // Our handle on zookeeper
@@ -79,7 +79,6 @@ public class ReplicationZookeeper extend
   private final Configuration conf;
   // Abortable
   private Abortable abortable;
-  private final ReplicationStateInterface replicationState;
   private final ReplicationPeers replicationPeers;
   private final ReplicationQueues replicationQueues;
 
@@ -95,8 +94,6 @@ public class ReplicationZookeeper extend
     this.conf = conf;
     this.zookeeper = zk;
     setZNodes(abortable);
-    this.replicationState = new ReplicationStateImpl(this.zookeeper, conf, abortable);
-    this.replicationState.init();
     // TODO This interface is no longer used by anyone using this constructor. When this
class goes
     // away, we will no longer have this null initialization business
     this.replicationQueues = null;
@@ -108,19 +105,16 @@ public class ReplicationZookeeper extend
    * Constructor used by region servers, connects to the peer cluster right away.
    *
    * @param server
-   * @param replicating    atomic boolean to start/stop replication
    * @throws IOException
    * @throws KeeperException
    */
-  public ReplicationZookeeper(final Server server, final AtomicBoolean replicating)
+  public ReplicationZookeeper(final Server server)
   throws IOException, KeeperException {
     super(server.getZooKeeper(), server.getConfiguration(), server);
     this.abortable = server;
     this.zookeeper = server.getZooKeeper();
     this.conf = server.getConfiguration();
     setZNodes(server);
-    this.replicationState = new ReplicationStateImpl(this.zookeeper, conf, server, replicating);
-    this.replicationState.init();
     this.replicationQueues = new ReplicationQueuesZKImpl(this.zookeeper, this.conf, server);
     this.replicationQueues.init(server.getServerName().toString());
     this.replicationPeers = new ReplicationPeersZKImpl(this.zookeeper, this.conf, server);
@@ -228,25 +222,6 @@ public class ReplicationZookeeper extend
   }
 
   /**
-   * Get the replication status of this cluster. If the state znode doesn't exist it will
also
-   * create it and set it true.
-   * @return returns true when it's enabled, else false
-   * @throws KeeperException
-   */
-  public boolean getReplication() throws KeeperException {
-    return this.replicationState.getState();
-  }
-
-  /**
-   * Set the new replication state for this cluster
-   * @param newState
-   * @throws KeeperException
-   */
-  public void setReplication(boolean newState) throws KeeperException {
-    this.replicationState.setState(newState);
-  }
-
-  /**
    * Add a new log to the list of hlogs in zookeeper
    * @param filename name of the hlog's znode
    * @param peerId name of the cluster's znode
@@ -388,9 +363,4 @@ public class ReplicationZookeeper extend
   public String getPeersZNode() {
     return peersZNode;
   }
-
-  @Override
-  public void close() throws IOException {
-    if (replicationState != null) replicationState.close();
-  }
 }

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplication.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplication.java?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplication.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplication.java
Tue Jul  9 19:59:39 2013
@@ -111,7 +111,6 @@ public class VerifyReplication {
           @Override
           public Void connect(HConnection conn) throws IOException {
             ZooKeeperWatcher localZKW = null;
-            ReplicationZookeeper zk = null;
             ReplicationPeer peer = null;
             try {
               localZKW = new ZooKeeperWatcher(
@@ -134,9 +133,6 @@ public class VerifyReplication {
               if (peer != null) {
                 peer.close();
               }
-              if (zk != null) {
-                zk.close();
-              }
               if (localZKW != null) {
                 localZKW.close();
               }

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/master/ReplicationLogCleaner.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/master/ReplicationLogCleaner.java?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/master/ReplicationLogCleaner.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/master/ReplicationLogCleaner.java
Tue Jul  9 19:59:39 2013
@@ -23,15 +23,12 @@ import org.apache.commons.logging.LogFac
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.Abortable;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.client.HConnectionManager;
 import org.apache.hadoop.hbase.master.cleaner.BaseLogCleanerDelegate;
 import org.apache.hadoop.hbase.replication.ReplicationQueuesClient;
 import org.apache.hadoop.hbase.replication.ReplicationQueuesClientZKImpl;
-import org.apache.hadoop.hbase.replication.ReplicationStateImpl;
-import org.apache.hadoop.hbase.replication.ReplicationStateInterface;
 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
 import org.apache.zookeeper.KeeperException;
 
@@ -49,7 +46,6 @@ public class ReplicationLogCleaner exten
   private static final Log LOG = LogFactory.getLog(ReplicationLogCleaner.class);
   private ZooKeeperWatcher zkw;
   private ReplicationQueuesClient replicationQueues;
-  private ReplicationStateInterface replicationState;
   private final Set<String> hlogs = new HashSet<String>();
   private boolean stopped = false;
   private boolean aborted;
@@ -58,15 +54,6 @@ public class ReplicationLogCleaner exten
   @Override
   public boolean isLogDeletable(FileStatus fStat) {
 
-    try {
-      if (!replicationState.getState()) {
-        return false;
-      }
-    } catch (KeeperException e) {
-      abort("Cannot get the state of replication", e);
-      return false;
-    }
-
     // all members of this class are null if replication is disabled, and we
     // return true since false would render the LogsCleaner useless
     if (this.getConf() == null) {
@@ -136,8 +123,6 @@ public class ReplicationLogCleaner exten
     try {
       this.zkw = new ZooKeeperWatcher(conf, "replicationLogCleaner", null);
       this.replicationQueues = new ReplicationQueuesClientZKImpl(zkw, conf, this);
-      this.replicationState = new ReplicationStateImpl(zkw, conf, this);
-      this.replicationState.init();
     } catch (KeeperException e) {
       LOG.error("Error while configuring " + this.getClass().getName(), e);
     } catch (IOException e) {
@@ -155,14 +140,6 @@ public class ReplicationLogCleaner exten
       LOG.info("Stopping " + this.zkw);
       this.zkw.close();
     }
-    if (this.replicationState != null) {
-      LOG.info("Stopping " + this.replicationState);
-      try {
-        this.replicationState.close();
-      } catch (IOException e) {
-        LOG.error("Error while stopping " + this.replicationState, e);
-      }
-    }
     // Not sure why we're deleting a connection that we never acquired or used
     HConnectionManager.deleteConnection(this.getConf());
   }

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/Replication.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/Replication.java?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/Replication.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/Replication.java
Tue Jul  9 19:59:39 2013
@@ -66,7 +66,6 @@ public class Replication implements WALA
       LogFactory.getLog(Replication.class);
   private boolean replication;
   private ReplicationSourceManager replicationManager;
-  private final AtomicBoolean replicating = new AtomicBoolean(true);
   private ReplicationZookeeper zkHelper;
   private ReplicationQueues replicationQueues;
   private Configuration conf;
@@ -108,17 +107,16 @@ public class Replication implements WALA
         .build());
     if (replication) {
       try {
-        this.zkHelper = new ReplicationZookeeper(server, this.replicating);
+        this.zkHelper = new ReplicationZookeeper(server);
         this.replicationQueues =
             new ReplicationQueuesZKImpl(server.getZooKeeper(), this.conf, this.server);
         this.replicationQueues.init(this.server.getServerName().toString());
       } catch (KeeperException ke) {
-        throw new IOException("Failed replication handler create " +
-           "(replicating=" + this.replicating, ke);
+        throw new IOException("Failed replication handler create", ke);
       }
       this.replicationManager =
-          new ReplicationSourceManager(zkHelper, replicationQueues, conf, this.server, fs,
-              this.replicating, logDir, oldLogDir);
+          new ReplicationSourceManager(zkHelper, replicationQueues, conf, this.server, fs,
logDir,
+              oldLogDir);
       this.statsThreadPeriod =
           this.conf.getInt("replication.stats.thread.period.seconds", 5 * 60);
       LOG.debug("ReplicationStatisticsThread " + this.statsThreadPeriod);

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java
Tue Jul  9 19:59:39 2013
@@ -112,9 +112,8 @@ public class ReplicationSink {
    }
 
   /**
-   * Replicate this array of entries directly into the local cluster using the native client.
-   * Only operates against raw protobuf type saving on a convertion from pb to pojo.
-   *
+   * Replicate this array of entries directly into the local cluster using the native client.
Only
+   * operates against raw protobuf type saving on a conversion from pb to pojo.
    * @param entries
    * @param cells
    * @throws IOException

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
Tue Jul  9 19:59:39 2013
@@ -93,8 +93,6 @@ public class ReplicationSource extends T
   // ratio of region servers to chose from a slave cluster
   private float ratio;
   private Random random;
-  // should we replicate or not?
-  private AtomicBoolean replicating;
   private ReplicationQueueInfo replicationQueueInfo;
   // id of the peer cluster this source replicates to
   private String peerId;
@@ -149,7 +147,6 @@ public class ReplicationSource extends T
    * @param fs file system to use
    * @param manager replication manager to ping to
    * @param stopper     the atomic boolean to use to stop the regionserver
-   * @param replicating the atomic boolean that starts/stops replication
    * @param peerClusterZnode the name of our znode
    * @throws IOException
    */
@@ -157,7 +154,6 @@ public class ReplicationSource extends T
                    final FileSystem fs,
                    final ReplicationSourceManager manager,
                    final Stoppable stopper,
-                   final AtomicBoolean replicating,
                    final String peerClusterZnode)
       throws IOException {
     this.stopper = stopper;
@@ -185,7 +181,6 @@ public class ReplicationSource extends T
     this.ratio = this.conf.getFloat("replication.source.ratio", 0.1f);
     this.currentPeers = new ArrayList<ServerName>();
     this.random = new Random();
-    this.replicating = replicating;
     this.manager = manager;
     this.sleepForRetries =
         this.conf.getLong("replication.source.sleepforretries", 1000);
@@ -417,9 +412,8 @@ public class ReplicationSource extends T
         removeNonReplicableEdits(entry);
         // Don't replicate catalog entries, if the WALEdit wasn't
         // containing anything to replicate and if we're currently not set to replicate
-        if (!(Bytes.equals(logKey.getTablename(), HConstants.ROOT_TABLE_NAME) ||
-            Bytes.equals(logKey.getTablename(), HConstants.META_TABLE_NAME)) &&
-            edit.size() != 0 && replicating.get()) {
+        if (!(Bytes.equals(logKey.getTablename(), HConstants.ROOT_TABLE_NAME) || Bytes.equals(
+          logKey.getTablename(), HConstants.META_TABLE_NAME)) && edit.size() != 0)
{
           // Only set the clusterId if is a local key.
           // This ensures that the originator sets the cluster id
           // and all replicas retain the initial cluster id.
@@ -714,8 +708,7 @@ public class ReplicationSource extends T
    * @return true if the peer is enabled, otherwise false
    */
   protected boolean isPeerEnabled() {
-    return this.replicating.get() &&
-        this.zkHelper.getPeerEnabled(this.peerId);
+    return this.zkHelper.getPeerEnabled(this.peerId);
   }
 
   /**

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java
Tue Jul  9 19:59:39 2013
@@ -19,7 +19,6 @@
 package org.apache.hadoop.hbase.replication.regionserver;
 
 import java.io.IOException;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
@@ -39,7 +38,6 @@ public interface ReplicationSourceInterf
    * @param fs the file system to use
    * @param manager the manager to use
    * @param stopper the stopper object for this region server
-   * @param replicating the status of the replication on this cluster
    * @param peerClusterId the id of the peer cluster
    * @throws IOException
    */
@@ -47,7 +45,6 @@ public interface ReplicationSourceInterf
                    final FileSystem fs,
                    final ReplicationSourceManager manager,
                    final Stoppable stopper,
-                   final AtomicBoolean replicating,
                    final String peerClusterId) throws IOException;
 
   /**

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java
(original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java
Tue Jul  9 19:59:39 2013
@@ -69,8 +69,6 @@ public class ReplicationSourceManager {
   private final List<ReplicationSourceInterface> sources;
   // List of all the sources we got from died RSs
   private final List<ReplicationSourceInterface> oldsources;
-  // Indicates if we are currently replicating
-  private final AtomicBoolean replicating;
   // Helper for zookeeper
   private final ReplicationZookeeper zkHelper;
   private final ReplicationQueues replicationQueues;
@@ -103,16 +101,13 @@ public class ReplicationSourceManager {
    * @param conf the configuration to use
    * @param stopper the stopper object for this region server
    * @param fs the file system to use
-   * @param replicating the status of the replication on this cluster
    * @param logDir the directory that contains all hlog directories of live RSs
    * @param oldLogDir the directory where old logs are archived
    */
   public ReplicationSourceManager(final ReplicationZookeeper zkHelper,
       final ReplicationQueues replicationQueues, final Configuration conf, final Stoppable
stopper,
-      final FileSystem fs, final AtomicBoolean replicating, final Path logDir,
-      final Path oldLogDir) {
+      final FileSystem fs, final Path logDir, final Path oldLogDir) {
     this.sources = new ArrayList<ReplicationSourceInterface>();
-    this.replicating = replicating;
     this.zkHelper = zkHelper;
     this.replicationQueues = replicationQueues;
     this.stopper = stopper;
@@ -206,8 +201,7 @@ public class ReplicationSourceManager {
    * @throws IOException
    */
   public ReplicationSourceInterface addSource(String id) throws IOException {
-    ReplicationSourceInterface src =
-        getReplicationSource(this.conf, this.fs, this, stopper, replicating, id);
+    ReplicationSourceInterface src = getReplicationSource(this.conf, this.fs, this, stopper,
id);
     synchronized (this.hlogsById) {
       this.sources.add(src);
       this.hlogsById.put(id, new TreeSet<String>());
@@ -260,10 +254,6 @@ public class ReplicationSourceManager {
   }
 
   void preLogRoll(Path newLog) throws IOException {
-    if (!this.replicating.get()) {
-      LOG.warn("Replication stopped, won't add new log");
-      return;
-    }
 
     synchronized (this.hlogsById) {
       String name = newLog.getName();
@@ -288,14 +278,9 @@ public class ReplicationSourceManager {
   }
 
   void postLogRoll(Path newLog) throws IOException {
-    if (!this.replicating.get()) {
-      LOG.warn("Replication stopped, won't add new log");
-      return;
-    }
-
     // This only updates the sources we own, not the recovered ones
     for (ReplicationSourceInterface source : this.sources) {
-      source.enqueueLog(newLog);    
+      source.enqueueLog(newLog);
     }
   }
 
@@ -313,7 +298,6 @@ public class ReplicationSourceManager {
    * @param fs the file system to use
    * @param manager the manager to use
    * @param stopper the stopper object for this region server
-   * @param replicating the status of the replication on this cluster
    * @param peerId the id of the peer cluster
    * @return the created source
    * @throws IOException
@@ -323,7 +307,6 @@ public class ReplicationSourceManager {
       final FileSystem fs,
       final ReplicationSourceManager manager,
       final Stoppable stopper,
-      final AtomicBoolean replicating,
       final String peerId) throws IOException {
     ReplicationSourceInterface src;
     try {
@@ -337,7 +320,7 @@ public class ReplicationSourceManager {
       src = new ReplicationSource();
 
     }
-    src.init(conf, fs, manager, stopper, replicating, peerId);
+    src.init(conf, fs, manager, stopper, peerId);
     return src;
   }
 
@@ -599,8 +582,8 @@ public class ReplicationSourceManager {
       for (Map.Entry<String, SortedSet<String>> entry : newQueues.entrySet())
{
         String peerId = entry.getKey();
         try {
-          ReplicationSourceInterface src = getReplicationSource(conf,
-              fs, ReplicationSourceManager.this, stopper, replicating, peerId);
+          ReplicationSourceInterface src =
+              getReplicationSource(conf, fs, ReplicationSourceManager.this, stopper, peerId);
           if (!zkHelper.getPeerClusters().contains((src.getPeerClusterId()))) {
             src.terminate("Recovered queue doesn't belong to any current peer");
             break;

Modified: hbase/trunk/hbase-server/src/main/javadoc/org/apache/hadoop/hbase/replication/package.html
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/javadoc/org/apache/hadoop/hbase/replication/package.html?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/javadoc/org/apache/hadoop/hbase/replication/package.html
(original)
+++ hbase/trunk/hbase-server/src/main/javadoc/org/apache/hadoop/hbase/replication/package.html
Tue Jul  9 19:59:39 2013
@@ -132,16 +132,6 @@ Choosing peer 10.10.1.49:62020</pre>
 
 In this case it indicates that 1 region server from the slave cluster
 was chosen for replication.<br><br>
-
-Should you want to stop the replication while the clusters are running, open
-the shell on the master cluster and issue this command:
-<pre>
-hbase(main):001:0> stop_replication</pre>
-
-Replication of already queued edits will still happen after you
-issued that command but new entries won't be. To start it back, simply replace
-"false" with "true" in the command. 
-
 <p>
 
 <a name="verify">

Modified: hbase/trunk/hbase-server/src/main/ruby/hbase/replication_admin.rb
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/ruby/hbase/replication_admin.rb?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/ruby/hbase/replication_admin.rb (original)
+++ hbase/trunk/hbase-server/src/main/ruby/hbase/replication_admin.rb Tue Jul  9 19:59:39
2013
@@ -65,17 +65,5 @@ module Hbase
     def disable_peer(id)
       @replication_admin.disablePeer(id)
     end
-
-    #----------------------------------------------------------------------------------------------
-    # Restart the replication, in an unknown state
-    def start_replication
-      @replication_admin.setReplicating(true)
-    end
-
-    #----------------------------------------------------------------------------------------------
-    # Kill switch for replication, stops all its features
-    def stop_replication
-      @replication_admin.setReplicating(false)
-    end
   end
 end

Modified: hbase/trunk/hbase-server/src/main/ruby/shell.rb
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/ruby/shell.rb?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/ruby/shell.rb (original)
+++ hbase/trunk/hbase-server/src/main/ruby/shell.rb Tue Jul  9 19:59:39 2013
@@ -297,15 +297,13 @@ Shell.load_command_group(
 Shell.load_command_group(
   'replication',
   :full_name => 'CLUSTER REPLICATION TOOLS',
-  :comment => "In order to use these tools, hbase.replication must be true. enabling/disabling
is currently unsupported",
+  :comment => "In order to use these tools, hbase.replication must be true.",
   :commands => %w[
     add_peer
     remove_peer
     list_peers
     enable_peer
     disable_peer
-    start_replication
-    stop_replication
   ]
 )
 

Modified: hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestLogsCleaner.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestLogsCleaner.java?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestLogsCleaner.java
(original)
+++ hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestLogsCleaner.java
Tue Jul  9 19:59:39 2013
@@ -68,8 +68,7 @@ public class TestLogsCleaner {
     conf.setBoolean(HConstants.REPLICATION_ENABLE_KEY, true);
     Replication.decorateMasterConfiguration(conf);
     Server server = new DummyServer();
-    ReplicationZookeeper zkHelper =
-        new ReplicationZookeeper(server, new AtomicBoolean(true));
+    ReplicationZookeeper zkHelper = new ReplicationZookeeper(server);
 
     Path oldLogDir = new Path(TEST_UTIL.getDataTestDir(),
         HConstants.HREGION_OLDLOGDIR_NAME);

Modified: hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java
(original)
+++ hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java
Tue Jul  9 19:59:39 2013
@@ -38,10 +38,8 @@ public class ReplicationSourceDummy impl
   Path currentPath;
 
   @Override
-  public void init(Configuration conf, FileSystem fs,
-                   ReplicationSourceManager manager, Stoppable stopper,
-                   AtomicBoolean replicating, String peerClusterId)
-      throws IOException {
+  public void init(Configuration conf, FileSystem fs, ReplicationSourceManager manager,
+      Stoppable stopper, String peerClusterId) throws IOException {
     this.manager = manager;
     this.peerClusterId = peerClusterId;
   }

Modified: hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationBase.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationBase.java?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationBase.java
(original)
+++ hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationBase.java
Tue Jul  9 19:59:39 2013
@@ -111,7 +111,6 @@ public class TestReplicationBase {
     zkw2 = new ZooKeeperWatcher(conf2, "cluster2", null, true);
 
     admin.addPeer("2", utility2.getClusterKey());
-    setIsReplication(true);
 
     LOG.info("Setup second Zk");
     CONF_WITH_LOCALFS = HBaseConfiguration.create(conf1);
@@ -134,12 +133,6 @@ public class TestReplicationBase {
     htable2 = new HTable(conf2, tableName);
   }
 
-  protected static void setIsReplication(boolean rep) throws Exception {
-    LOG.info("Set rep " + rep);
-    admin.setReplicating(rep);
-    Thread.sleep(SLEEP_TIME);
-  }
-
   /**
    * @throws java.lang.Exception
    */

Modified: hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationSmallTests.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationSmallTests.java?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationSmallTests.java
(original)
+++ hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationSmallTests.java
Tue Jul  9 19:59:39 2013
@@ -256,75 +256,6 @@ public class TestReplicationSmallTests e
   }
 
   /**
-   * Test stopping replication, trying to insert, make sure nothing's
-   * replicated, enable it, try replicating and it should work
-   * @throws Exception
-   */
-  @Test(timeout=300000)
-  public void testStartStop() throws Exception {
-
-    // Test stopping replication
-    setIsReplication(false);
-
-    Put put = new Put(Bytes.toBytes("stop start"));
-    put.add(famName, row, row);
-    htable1.put(put);
-
-    Get get = new Get(Bytes.toBytes("stop start"));
-    for (int i = 0; i < NB_RETRIES; i++) {
-      if (i==NB_RETRIES-1) {
-        break;
-      }
-      Result res = htable2.get(get);
-      if(res.size() >= 1) {
-        fail("Replication wasn't stopped");
-
-      } else {
-        LOG.info("Row not replicated, let's wait a bit more...");
-        Thread.sleep(SLEEP_TIME);
-      }
-    }
-
-    // Test restart replication
-    setIsReplication(true);
-
-    htable1.put(put);
-
-    for (int i = 0; i < NB_RETRIES; i++) {
-      if (i==NB_RETRIES-1) {
-        fail("Waited too much time for put replication");
-      }
-      Result res = htable2.get(get);
-      if(res.size() == 0) {
-        LOG.info("Row not available");
-        Thread.sleep(SLEEP_TIME);
-      } else {
-        assertArrayEquals(res.value(), row);
-        break;
-      }
-    }
-
-    put = new Put(Bytes.toBytes("do not rep"));
-    put.add(noRepfamName, row, row);
-    htable1.put(put);
-
-    get = new Get(Bytes.toBytes("do not rep"));
-    for (int i = 0; i < NB_RETRIES; i++) {
-      if (i == NB_RETRIES-1) {
-        break;
-      }
-      Result res = htable2.get(get);
-      if (res.size() >= 1) {
-        fail("Not supposed to be replicated");
-      } else {
-        LOG.info("Row not replicated, let's wait a bit more...");
-        Thread.sleep(SLEEP_TIME);
-      }
-    }
-
-  }
-
-  /**
    * Test disable/enable replication, trying to insert, make sure nothing's
    * replicated, enable it, the insert should be replicated
    *

Modified: hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationStateZKImpl.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationStateZKImpl.java?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationStateZKImpl.java
(original)
+++ hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationStateZKImpl.java
Tue Jul  9 19:59:39 2013
@@ -69,10 +69,6 @@ public class TestReplicationStateZKImpl 
     Configuration testConf = new Configuration(conf);
     testConf.set(HConstants.ZOOKEEPER_ZNODE_PARENT, baseZKNode);
     ZooKeeperWatcher zkw1 = new ZooKeeperWatcher(testConf, "test1", null);
-    ReplicationStateInterface rsi = new ReplicationStateImpl(zkw1, testConf, zkw1);
-    rsi.init();
-    rsi.setState(true);
-    rsi.close();
     String fakeRs = ZKUtil.joinZNode(zkw1.rsZNode, "hostname1.example.org:1234");
     ZKUtil.createWithParents(zkw1, fakeRs);
     ZKClusterId.setClusterId(zkw1, new ClusterId());

Modified: hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationZookeeper.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationZookeeper.java?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationZookeeper.java
(original)
+++ hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationZookeeper.java
Tue Jul  9 19:59:39 2013
@@ -62,7 +62,7 @@ public class TestReplicationZookeeper {
     conf = utility.getConfiguration();
     zkw = HBaseTestingUtility.getZooKeeperWatcher(utility);
     DummyServer server = new DummyServer();
-    repZk = new ReplicationZookeeper(server, new AtomicBoolean());
+    repZk = new ReplicationZookeeper(server);
     slaveClusterKey = conf.get(HConstants.ZOOKEEPER_QUORUM) + ":" +
       conf.get("hbase.zookeeper.property.clientPort") + ":/1";
     String replicationZNodeName = conf.get("zookeeper.znode.replication", "replication");

Modified: hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceManager.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceManager.java?rev=1501503&r1=1501502&r2=1501503&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceManager.java
(original)
+++ hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceManager.java
Tue Jul  9 19:59:39 2013
@@ -28,7 +28,6 @@ import java.util.List;
 import java.util.SortedMap;
 import java.util.SortedSet;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -227,8 +226,7 @@ public class TestReplicationSourceManage
     LOG.debug("testNodeFailoverWorkerCopyQueuesFromRSUsingMulti");
     conf.setBoolean(HConstants.ZOOKEEPER_USEMULTI, true);
     final Server server = new DummyServer("hostname0.example.org");
-    AtomicBoolean replicating = new AtomicBoolean(true);
-    ReplicationZookeeper rz = new ReplicationZookeeper(server, replicating);
+    ReplicationZookeeper rz = new ReplicationZookeeper(server);
     // populate some znodes in the peer znode
     files.add("log1");
     files.add("log2");
@@ -260,8 +258,6 @@ public class TestReplicationSourceManage
     populatedMap += w1.isLogZnodesMapPopulated() + w2.isLogZnodesMapPopulated()
         + w3.isLogZnodesMapPopulated();
     assertEquals(1, populatedMap);
-    // close out the resources.
-    rz.close();
     server.abort("", null);
   }
 
@@ -270,8 +266,7 @@ public class TestReplicationSourceManage
     LOG.debug("testNodeFailoverDeadServerParsing");
     conf.setBoolean(HConstants.ZOOKEEPER_USEMULTI, true);
     final Server server = new DummyServer("ec2-54-234-230-108.compute-1.amazonaws.com");
-    AtomicBoolean replicating = new AtomicBoolean(true);
-    ReplicationZookeeper rz = new ReplicationZookeeper(server, replicating);
+    ReplicationZookeeper rz = new ReplicationZookeeper(server);
     // populate some znodes in the peer znode
     files.add("log1");
     files.add("log2");
@@ -285,16 +280,13 @@ public class TestReplicationSourceManage
     Server s3 = new DummyServer("ec2-23-20-187-167.compute-1.amazonaws.com");
 
     // simulate three servers fail sequentially
-    ReplicationZookeeper rz1 = new ReplicationZookeeper(s1, new AtomicBoolean(true));
+    ReplicationZookeeper rz1 = new ReplicationZookeeper(s1);
     SortedMap<String, SortedSet<String>> testMap =
         rz1.claimQueues(server.getServerName().getServerName());
-    rz1.close();
-    ReplicationZookeeper rz2 = new ReplicationZookeeper(s2, new AtomicBoolean(true));
+    ReplicationZookeeper rz2 = new ReplicationZookeeper(s2);
     testMap = rz2.claimQueues(s1.getServerName().getServerName());
-    rz2.close();
-    ReplicationZookeeper rz3 = new ReplicationZookeeper(s3, new AtomicBoolean(true));
+    ReplicationZookeeper rz3 = new ReplicationZookeeper(s3);
     testMap = rz3.claimQueues(s2.getServerName().getServerName());
-    rz3.close();
 
     ReplicationQueueInfo replicationQueueInfo = new ReplicationQueueInfo(testMap.firstKey());
     List<String> result = replicationQueueInfo.getDeadRegionServers();
@@ -303,9 +295,7 @@ public class TestReplicationSourceManage
     assertTrue(result.contains(server.getServerName().getServerName()));
     assertTrue(result.contains(s1.getServerName().getServerName()));
     assertTrue(result.contains(s2.getServerName().getServerName()));
-    
-    // close out the resources.
-    rz.close();
+
     server.abort("", null);
   }
   
@@ -319,14 +309,13 @@ public class TestReplicationSourceManage
     public DummyNodeFailoverWorker(String znode, Server s) throws Exception {
       this.deadRsZnode = znode;
       this.server = s;
-      rz = new ReplicationZookeeper(server, new AtomicBoolean(true));
+      rz = new ReplicationZookeeper(server);
     }
 
     @Override
     public void run() {
       try {
         logZnodesMap = rz.claimQueues(deadRsZnode);
-        rz.close();
         server.abort("Done with testing", null);
       } catch (Exception e) {
         LOG.error("Got exception while running NodeFailoverWorker", e);



Mime
View raw message