001/*
002 *
003 * Licensed to the Apache Software Foundation (ASF) under one
004 * or more contributor license agreements.  See the NOTICE file
005 * distributed with this work for additional information
006 * regarding copyright ownership.  The ASF licenses this file
007 * to you under the Apache License, Version 2.0 (the
008 * "License"); you may not use this file except in compliance
009 * with the License.  You may obtain a copy of the License at
010 *
011 *     http://www.apache.org/licenses/LICENSE-2.0
012 *
013 * Unless required by applicable law or agreed to in writing, software
014 * distributed under the License is distributed on an "AS IS" BASIS,
015 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
016 * See the License for the specific language governing permissions and
017 * limitations under the License.
018 */
019package org.apache.hadoop.hbase.replication.regionserver;
020
021import java.io.IOException;
022import java.util.HashMap;
023import java.util.List;
024import java.util.Map;
025import java.util.UUID;
026
027import org.apache.hadoop.conf.Configuration;
028import org.apache.hadoop.fs.FileSystem;
029import org.apache.hadoop.fs.Path;
030import org.apache.hadoop.hbase.Server;
031import org.apache.hadoop.hbase.ServerName;
032import org.apache.hadoop.hbase.TableName;
033import org.apache.hadoop.hbase.replication.ReplicationEndpoint;
034import org.apache.hadoop.hbase.replication.ReplicationException;
035import org.apache.hadoop.hbase.replication.ReplicationPeer;
036import org.apache.hadoop.hbase.replication.ReplicationQueueStorage;
037import org.apache.hadoop.hbase.util.Pair;
038import org.apache.hadoop.hbase.wal.WAL.Entry;
039import org.apache.yetus.audience.InterfaceAudience;
040
041/**
042 * Interface that defines a replication source
043 */
044@InterfaceAudience.Private
045public interface ReplicationSourceInterface {
046
047  /**
048   * Initializer for the source
049   * @param conf the configuration to use
050   * @param fs the file system to use
051   * @param manager the manager to use
052   * @param server the server for this region server
053   */
054  void init(Configuration conf, FileSystem fs, ReplicationSourceManager manager,
055      ReplicationQueueStorage queueStorage, ReplicationPeer replicationPeer, Server server,
056      String queueId, UUID clusterId, WALFileLengthProvider walFileLengthProvider,
057      MetricsSource metrics) throws IOException;
058
059  /**
060   * Add a log to the list of logs to replicate
061   * @param log path to the log to replicate
062   */
063  void enqueueLog(Path log);
064
065  /**
066   * Add hfile names to the queue to be replicated.
067   * @param tableName Name of the table these files belongs to
068   * @param family Name of the family these files belong to
069   * @param pairs list of pairs of { HFile location in staging dir, HFile path in region dir which
070   *          will be added in the queue for replication}
071   * @throws ReplicationException If failed to add hfile references
072   */
073  void addHFileRefs(TableName tableName, byte[] family, List<Pair<Path, Path>> pairs)
074      throws ReplicationException;
075
076  /**
077   * Start the replication
078   */
079  void startup();
080
081  /**
082   * End the replication
083   * @param reason why it's terminating
084   */
085  void terminate(String reason);
086
087  /**
088   * End the replication
089   * @param reason why it's terminating
090   * @param cause the error that's causing it
091   */
092  void terminate(String reason, Exception cause);
093
094  /**
095   * End the replication
096   * @param reason why it's terminating
097   * @param cause the error that's causing it
098   * @param clearMetrics removes all metrics about this Source
099   */
100  void terminate(String reason, Exception cause, boolean clearMetrics);
101
102  /**
103   * Get the current log that's replicated
104   * @return the current log
105   */
106  Path getCurrentPath();
107
108  /**
109   * Get the queue id that the source is replicating to
110   *
111   * @return queue id
112   */
113  String getQueueId();
114
115  /**
116   * Get the id that the source is replicating to.
117   *
118   * @return peer id
119   */
120  String getPeerId();
121
122  /**
123   * Get a string representation of the current statistics
124   * for this source
125   * @return printable stats
126   */
127  String getStats();
128
129  /**
130   * @return peer enabled or not
131   */
132  boolean isPeerEnabled();
133
134  /**
135   * @return active or not
136   */
137  boolean isSourceActive();
138
139  /**
140   * @return metrics of this replication source
141   */
142  MetricsSource getSourceMetrics();
143
144  /**
145   * @return the replication endpoint used by this replication source
146   */
147  ReplicationEndpoint getReplicationEndpoint();
148
149  /**
150   * @return the replication source manager
151   */
152  ReplicationSourceManager getSourceManager();
153
154  /**
155   * @return the wal file length provider
156   */
157  WALFileLengthProvider getWALFileLengthProvider();
158
159  /**
160   * Try to throttle when the peer config with a bandwidth
161   * @param batchSize entries size will be pushed
162   * @throws InterruptedException
163   */
164  void tryThrottle(int batchSize) throws InterruptedException;
165
166  /**
167   * Call this after the shipper thread ship some entries to peer cluster.
168   * @param entries pushed
169   * @param batchSize entries size pushed
170   */
171  void postShipEdits(List<Entry> entries, int batchSize);
172
173  /**
174   * The queue of WALs only belong to one region server. This will return the server name which all
175   * WALs belong to.
176   * @return the server name which all WALs belong to
177   */
178  ServerName getServerWALsBelongTo();
179
180  /**
181   * get the stat of replication for each wal group.
182   * @return stat of replication
183   */
184  default Map<String, ReplicationStatus> getWalGroupStatus() {
185    return new HashMap<>();
186  }
187
188  /**
189   * @return whether this is a replication source for recovery.
190   */
191  default boolean isRecovered() {
192    return false;
193  }
194}