001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.replication.regionserver;
019
020import java.io.IOException;
021import org.apache.hadoop.conf.Configuration;
022import org.apache.hadoop.conf.Configured;
023import org.apache.hadoop.fs.FileSystem;
024import org.apache.hadoop.fs.Path;
025import org.apache.hadoop.hbase.Abortable;
026import org.apache.hadoop.hbase.ChoreService;
027import org.apache.hadoop.hbase.CoordinatedStateManager;
028import org.apache.hadoop.hbase.HBaseConfiguration;
029import org.apache.hadoop.hbase.HConstants;
030import org.apache.hadoop.hbase.Server;
031import org.apache.hadoop.hbase.ServerName;
032import org.apache.hadoop.hbase.client.ClusterConnection;
033import org.apache.hadoop.hbase.client.Connection;
034import org.apache.hadoop.hbase.util.CommonFSUtils;
035import org.apache.hadoop.hbase.wal.WALFactory;
036import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
037import org.apache.hadoop.util.Tool;
038import org.apache.hadoop.util.ToolRunner;
039import org.apache.yetus.audience.InterfaceAudience;
040
041/**
042 * In a scenario of Replication based Disaster/Recovery, when hbase Master-Cluster crashes, this
043 * tool is used to sync-up the delta from Master to Slave using the info from ZooKeeper. The tool
044 * will run on Master-Cluser, and assume ZK, Filesystem and NetWork still available after hbase
045 * crashes
046 *
047 * <pre>
048 * hbase org.apache.hadoop.hbase.replication.regionserver.ReplicationSyncUp
049 * </pre>
050 */
051@InterfaceAudience.Private
052public class ReplicationSyncUp extends Configured implements Tool {
053
054  private static final long SLEEP_TIME = 10000;
055
056  /**
057   * Main program
058   */
059  public static void main(String[] args) throws Exception {
060    int ret = ToolRunner.run(HBaseConfiguration.create(), new ReplicationSyncUp(), args);
061    System.exit(ret);
062  }
063
064  @Override
065  public int run(String[] args) throws Exception {
066    Abortable abortable = new Abortable() {
067      @Override
068      public void abort(String why, Throwable e) {
069      }
070
071      @Override
072      public boolean isAborted() {
073        return false;
074      }
075    };
076    Configuration conf = getConf();
077    try (ZKWatcher zkw =
078      new ZKWatcher(conf, "syncupReplication" + System.currentTimeMillis(), abortable, true)) {
079      Path walRootDir = CommonFSUtils.getWALRootDir(conf);
080      FileSystem fs = CommonFSUtils.getWALFileSystem(conf);
081      Path oldLogDir = new Path(walRootDir, HConstants.HREGION_OLDLOGDIR_NAME);
082      Path logDir = new Path(walRootDir, HConstants.HREGION_LOGDIR_NAME);
083
084      System.out.println("Start Replication Server start");
085      Replication replication = new Replication();
086      replication.initialize(new DummyServer(zkw), fs, logDir, oldLogDir,
087        new WALFactory(conf, "test", null));
088      ReplicationSourceManager manager = replication.getReplicationManager();
089      manager.init().get();
090      while (manager.activeFailoverTaskCount() > 0) {
091        Thread.sleep(SLEEP_TIME);
092      }
093      while (manager.getOldSources().size() > 0) {
094        Thread.sleep(SLEEP_TIME);
095      }
096      manager.join();
097    } catch (InterruptedException e) {
098      System.err.println("didn't wait long enough:" + e);
099      return -1;
100    }
101    return 0;
102  }
103
104  class DummyServer implements Server {
105    String hostname;
106    ZKWatcher zkw;
107
108    DummyServer(ZKWatcher zkw) {
109      // a unique name in case the first run fails
110      hostname = System.currentTimeMillis() + ".SyncUpTool.replication.org";
111      this.zkw = zkw;
112    }
113
114    DummyServer(String hostname) {
115      this.hostname = hostname;
116    }
117
118    @Override
119    public Configuration getConfiguration() {
120      return getConf();
121    }
122
123    @Override
124    public ZKWatcher getZooKeeper() {
125      return zkw;
126    }
127
128    @Override
129    public CoordinatedStateManager getCoordinatedStateManager() {
130      return null;
131    }
132
133    @Override
134    public ServerName getServerName() {
135      return ServerName.valueOf(hostname, 1234, 1L);
136    }
137
138    @Override
139    public void abort(String why, Throwable e) {
140    }
141
142    @Override
143    public boolean isAborted() {
144      return false;
145    }
146
147    @Override
148    public void stop(String why) {
149    }
150
151    @Override
152    public boolean isStopped() {
153      return false;
154    }
155
156    @Override
157    public ClusterConnection getConnection() {
158      return null;
159    }
160
161    @Override
162    public ChoreService getChoreService() {
163      return null;
164    }
165
166    @Override
167    public ClusterConnection getClusterConnection() {
168      return null;
169    }
170
171    @Override
172    public FileSystem getFileSystem() {
173      return null;
174    }
175
176    @Override
177    public boolean isStopping() {
178      return false;
179    }
180
181    @Override
182    public Connection createConnection(Configuration conf) throws IOException {
183      return null;
184    }
185  }
186}