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;
019
020import java.io.IOException;
021import java.util.Collection;
022import java.util.List;
023import java.util.Map;
024import java.util.Set;
025import org.apache.hadoop.conf.Configuration;
026import org.apache.hadoop.fs.FileSystem;
027import org.apache.hadoop.fs.Path;
028import org.apache.hadoop.hbase.HConstants;
029import org.apache.hadoop.hbase.TableName;
030import org.apache.yetus.audience.InterfaceAudience;
031import org.slf4j.Logger;
032import org.slf4j.LoggerFactory;
033
034/**
035 * Helper class for replication.
036 */
037@InterfaceAudience.Private
038public final class ReplicationUtils {
039
040  private static final Logger LOG = LoggerFactory.getLogger(ReplicationUtils.class);
041
042  public static final String REPLICATION_ATTR_NAME = "__rep__";
043
044  public static final String REMOTE_WAL_DIR_NAME = "remoteWALs";
045
046  public static final String SYNC_WAL_SUFFIX = ".syncrep";
047
048  public static final String REMOTE_WAL_REPLAY_SUFFIX = "-replay";
049
050  public static final String REMOTE_WAL_SNAPSHOT_SUFFIX = "-snapshot";
051
052  // This is used for copying sync replication log from local to remote and overwrite the old one
053  // since some FileSystem implementation may not support atomic rename.
054  public static final String RENAME_WAL_SUFFIX = ".ren";
055
056  public static final String LEGACY_REGION_REPLICATION_ENDPOINT_NAME =
057    "org.apache.hadoop.hbase.replication.regionserver.RegionReplicaReplicationEndpoint";
058
059  public static final String OFFSET_UPDATE_INTERVAL_MS_KEY =
060    "hbase.replication.shipper.offset.update.interval.ms";
061
062  public static final String OFFSET_UPDATE_SIZE_THRESHOLD_KEY =
063    "hbase.replication.shipper.offset.update.size.threshold";
064
065  private ReplicationUtils() {
066  }
067
068  private static boolean isCollectionEqual(Collection<String> c1, Collection<String> c2) {
069    if (c1 == null) {
070      return c2 == null;
071    }
072    if (c2 == null) {
073      return false;
074    }
075    return c1.size() == c2.size() && c1.containsAll(c2);
076  }
077
078  private static boolean isNamespacesEqual(Set<String> ns1, Set<String> ns2) {
079    return isCollectionEqual(ns1, ns2);
080  }
081
082  private static boolean isTableCFsEqual(Map<TableName, List<String>> tableCFs1,
083    Map<TableName, List<String>> tableCFs2) {
084    if (tableCFs1 == null) {
085      return tableCFs2 == null;
086    }
087    if (tableCFs2 == null) {
088      return false;
089    }
090    if (tableCFs1.size() != tableCFs2.size()) {
091      return false;
092    }
093    for (Map.Entry<TableName, List<String>> entry1 : tableCFs1.entrySet()) {
094      TableName table = entry1.getKey();
095      if (!tableCFs2.containsKey(table)) {
096        return false;
097      }
098      List<String> cfs1 = entry1.getValue();
099      List<String> cfs2 = tableCFs2.get(table);
100      if (!isCollectionEqual(cfs1, cfs2)) {
101        return false;
102      }
103    }
104    return true;
105  }
106
107  public static boolean isNamespacesAndTableCFsEqual(ReplicationPeerConfig rpc1,
108    ReplicationPeerConfig rpc2) {
109    if (rpc1.replicateAllUserTables() != rpc2.replicateAllUserTables()) {
110      return false;
111    }
112    if (rpc1.replicateAllUserTables()) {
113      return isNamespacesEqual(rpc1.getExcludeNamespaces(), rpc2.getExcludeNamespaces())
114        && isTableCFsEqual(rpc1.getExcludeTableCFsMap(), rpc2.getExcludeTableCFsMap());
115    } else {
116      return isNamespacesEqual(rpc1.getNamespaces(), rpc2.getNamespaces())
117        && isTableCFsEqual(rpc1.getTableCFsMap(), rpc2.getTableCFsMap());
118    }
119  }
120
121  /**
122   * @param c Configuration to look at
123   * @return True if replication for bulk load data is enabled.
124   */
125  public static boolean isReplicationForBulkLoadDataEnabled(final Configuration c) {
126    return c.getBoolean(HConstants.REPLICATION_BULKLOAD_ENABLE_KEY,
127      HConstants.REPLICATION_BULKLOAD_ENABLE_DEFAULT);
128  }
129
130  public static FileSystem getRemoteWALFileSystem(Configuration conf, String remoteWALDir)
131    throws IOException {
132    return new Path(remoteWALDir).getFileSystem(conf);
133  }
134
135  public static Path getPeerRemoteWALDir(String remoteWALDir, String peerId) {
136    return new Path(remoteWALDir, peerId);
137  }
138
139  public static Path getPeerRemoteWALDir(Path remoteWALDir, String peerId) {
140    return new Path(remoteWALDir, peerId);
141  }
142
143  public static Path getPeerReplayWALDir(Path remoteWALDir, String peerId) {
144    return getPeerRemoteWALDir(remoteWALDir, peerId).suffix(REMOTE_WAL_REPLAY_SUFFIX);
145  }
146
147  public static Path getPeerSnapshotWALDir(String remoteWALDir, String peerId) {
148    return getPeerRemoteWALDir(remoteWALDir, peerId).suffix(REMOTE_WAL_SNAPSHOT_SUFFIX);
149  }
150
151  public static Path getPeerSnapshotWALDir(Path remoteWALDir, String peerId) {
152    return getPeerRemoteWALDir(remoteWALDir, peerId).suffix(REMOTE_WAL_SNAPSHOT_SUFFIX);
153  }
154
155  /**
156   * Do the sleeping logic
157   * @param msg                  Why we sleep
158   * @param sleepForRetries      the base sleep time.
159   * @param sleepMultiplier      by how many times the default sleeping time is augmented
160   * @param maxRetriesMultiplier the max retry multiplier
161   * @return True if <code>sleepMultiplier</code> is &lt; <code>maxRetriesMultiplier</code>
162   */
163  public static boolean sleepForRetries(String msg, long sleepForRetries, int sleepMultiplier,
164    int maxRetriesMultiplier) {
165    try {
166      LOG.trace("{}, sleeping {} times {}", msg, sleepForRetries, sleepMultiplier);
167      Thread.sleep(sleepForRetries * sleepMultiplier);
168    } catch (InterruptedException e) {
169      LOG.debug("Interrupted while sleeping between retries");
170      Thread.currentThread().interrupt();
171    }
172    return sleepMultiplier < maxRetriesMultiplier;
173  }
174
175  /**
176   * Get the adaptive timeout value when performing a retry
177   */
178  public static int getAdaptiveTimeout(final int initialValue, final int retries) {
179    int ntries = retries;
180    if (ntries >= HConstants.RETRY_BACKOFF.length) {
181      ntries = HConstants.RETRY_BACKOFF.length - 1;
182    }
183    if (ntries < 0) {
184      ntries = 0;
185    }
186    return initialValue * HConstants.RETRY_BACKOFF[ntries];
187  }
188}