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.EOFException;
021import java.io.IOException;
022import java.util.ArrayList;
023import java.util.List;
024import java.util.concurrent.locks.Lock;
025import org.apache.hadoop.conf.Configuration;
026import org.apache.hadoop.fs.FileSystem;
027import org.apache.hadoop.fs.Path;
028import org.apache.hadoop.hbase.CellScanner;
029import org.apache.hadoop.hbase.CellUtil;
030import org.apache.hadoop.hbase.executor.EventType;
031import org.apache.hadoop.hbase.procedure2.RSProcedureCallable;
032import org.apache.hadoop.hbase.protobuf.ReplicationProtbufUtil;
033import org.apache.hadoop.hbase.regionserver.HRegionServer;
034import org.apache.hadoop.hbase.regionserver.wal.WALUtil;
035import org.apache.hadoop.hbase.util.KeyLocker;
036import org.apache.hadoop.hbase.util.Pair;
037import org.apache.hadoop.hbase.util.RecoverLeaseFSUtils;
038import org.apache.hadoop.hbase.wal.WAL.Entry;
039import org.apache.hadoop.hbase.wal.WAL.Reader;
040import org.apache.hadoop.hbase.wal.WALEdit;
041import org.apache.hadoop.hbase.wal.WALFactory;
042import org.apache.yetus.audience.InterfaceAudience;
043import org.slf4j.Logger;
044import org.slf4j.LoggerFactory;
045
046import org.apache.hbase.thirdparty.com.google.protobuf.InvalidProtocolBufferException;
047
048import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos;
049import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ReplicateWALEntryRequest;
050import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProcedureProtos.ReplaySyncReplicationWALParameter;
051
052/**
053 * This callable executed at RS side to replay sync replication wal.
054 */
055@InterfaceAudience.Private
056public class ReplaySyncReplicationWALCallable implements RSProcedureCallable {
057
058  private static final Logger LOG = LoggerFactory.getLogger(ReplaySyncReplicationWALCallable.class);
059
060  private static final String REPLAY_SYNC_REPLICATION_WAL_BATCH_SIZE =
061      "hbase.replay.sync.replication.wal.batch.size";
062
063  private static final long DEFAULT_REPLAY_SYNC_REPLICATION_WAL_BATCH_SIZE = 8 * 1024 * 1024;
064
065  private HRegionServer rs;
066
067  private FileSystem fs;
068
069  private Configuration conf;
070
071  private String peerId;
072
073  private List<String> wals = new ArrayList<>();
074
075  private Exception initError;
076
077  private long batchSize;
078
079  private final KeyLocker<String> peersLock = new KeyLocker<>();
080
081  @Override
082  public Void call() throws Exception {
083    if (initError != null) {
084      throw initError;
085    }
086    LOG.info("Received a replay sync replication wals {} event, peerId={}", wals, peerId);
087    if (rs.getReplicationSinkService() != null) {
088      Lock peerLock = peersLock.acquireLock(wals.get(0));
089      try {
090        for (String wal : wals) {
091          replayWAL(wal);
092        }
093      } finally {
094        peerLock.unlock();
095      }
096    }
097    return null;
098  }
099
100  @Override
101  public void init(byte[] parameter, HRegionServer rs) {
102    this.rs = rs;
103    this.fs = rs.getWALFileSystem();
104    this.conf = rs.getConfiguration();
105    try {
106      ReplaySyncReplicationWALParameter param =
107          ReplaySyncReplicationWALParameter.parseFrom(parameter);
108      this.peerId = param.getPeerId();
109      param.getWalList().forEach(this.wals::add);
110      this.batchSize = rs.getConfiguration().getLong(REPLAY_SYNC_REPLICATION_WAL_BATCH_SIZE,
111        DEFAULT_REPLAY_SYNC_REPLICATION_WAL_BATCH_SIZE);
112    } catch (InvalidProtocolBufferException e) {
113      initError = e;
114    }
115  }
116
117  @Override
118  public EventType getEventType() {
119    return EventType.RS_REPLAY_SYNC_REPLICATION_WAL;
120  }
121
122  private void replayWAL(String wal) throws IOException {
123    try (Reader reader = getReader(wal)) {
124      List<Entry> entries = readWALEntries(reader);
125      while (!entries.isEmpty()) {
126        Pair<AdminProtos.ReplicateWALEntryRequest, CellScanner> pair = ReplicationProtbufUtil
127            .buildReplicateWALEntryRequest(entries.toArray(new Entry[entries.size()]));
128        ReplicateWALEntryRequest request = pair.getFirst();
129        rs.getReplicationSinkService().replicateLogEntries(request.getEntryList(),
130            pair.getSecond(), request.getReplicationClusterId(),
131            request.getSourceBaseNamespaceDirPath(), request.getSourceHFileArchiveDirPath());
132        // Read next entries.
133        entries = readWALEntries(reader);
134      }
135    }
136  }
137
138  private Reader getReader(String wal) throws IOException {
139    Path path = new Path(rs.getWALRootDir(), wal);
140    long length = rs.getWALFileSystem().getFileStatus(path).getLen();
141    try {
142      RecoverLeaseFSUtils.recoverFileLease(fs, path, conf);
143      return WALFactory.createReader(rs.getWALFileSystem(), path, rs.getConfiguration());
144    } catch (EOFException e) {
145      if (length <= 0) {
146        LOG.warn("File is empty. Could not open {} for reading because {}", path, e);
147        return null;
148      }
149      throw e;
150    }
151  }
152
153  // return whether we should include this entry.
154  private boolean filter(Entry entry) {
155    WALEdit edit = entry.getEdit();
156    WALUtil.filterCells(edit, c -> CellUtil.matchingFamily(c, WALEdit.METAFAMILY) ? null : c);
157    return !edit.isEmpty();
158  }
159
160  private List<Entry> readWALEntries(Reader reader) throws IOException {
161    List<Entry> entries = new ArrayList<>();
162    if (reader == null) {
163      return entries;
164    }
165    long size = 0;
166    for (;;) {
167      Entry entry = reader.next();
168      if (entry == null) {
169        break;
170      }
171      if (filter(entry)) {
172        entries.add(entry);
173        size += entry.getEdit().heapSize();
174        if (size > batchSize) {
175          break;
176        }
177      }
178    }
179    return entries;
180  }
181}