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 java.util.List;
022import java.util.Optional;
023import org.apache.hadoop.conf.Configuration;
024import org.apache.hadoop.fs.Path;
025import org.apache.hadoop.hbase.HConstants;
026import org.apache.hadoop.hbase.TableName;
027import org.apache.hadoop.hbase.coprocessor.CoreCoprocessor;
028import org.apache.hadoop.hbase.coprocessor.HasRegionServerServices;
029import org.apache.hadoop.hbase.coprocessor.ObserverContext;
030import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
031import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
032import org.apache.hadoop.hbase.coprocessor.RegionObserver;
033import org.apache.hadoop.hbase.regionserver.HRegionServer;
034import org.apache.hadoop.hbase.regionserver.RegionServerServices;
035import org.apache.hadoop.hbase.util.Bytes;
036import org.apache.hadoop.hbase.util.Pair;
037import org.apache.yetus.audience.InterfaceAudience;
038import org.slf4j.Logger;
039import org.slf4j.LoggerFactory;
040
041/**
042 * An Observer to add HFile References to replication queue.
043 */
044@CoreCoprocessor
045@InterfaceAudience.Private
046public class ReplicationObserver implements RegionCoprocessor, RegionObserver {
047  private static final Logger LOG = LoggerFactory.getLogger(ReplicationObserver.class);
048
049  @Override
050  public Optional<RegionObserver> getRegionObserver() {
051    return Optional.of(this);
052  }
053
054  @Override
055  @edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "NP_NULL_ON_SOME_PATH",
056      justification = "NPE should never happen; if it does it is a bigger issue")
057  public void preCommitStoreFile(final ObserverContext<? extends RegionCoprocessorEnvironment> ctx,
058    final byte[] family, final List<Pair<Path, Path>> pairs) throws IOException {
059    RegionCoprocessorEnvironment env = ctx.getEnvironment();
060    Configuration c = env.getConfiguration();
061    if (
062      pairs == null || pairs.isEmpty()
063        || !c.getBoolean(HConstants.REPLICATION_BULKLOAD_ENABLE_KEY,
064          HConstants.REPLICATION_BULKLOAD_ENABLE_DEFAULT)
065    ) {
066      LOG.debug("Skipping recording bulk load entries in preCommitStoreFile for bulkloaded "
067        + "data replication.");
068      return;
069    }
070    TableName tableName = env.getRegionInfo().getTable();
071    if (
072      env.getRegion().getTableDescriptor().getColumnFamily(family).getScope()
073          != HConstants.REPLICATION_SCOPE_GLOBAL
074    ) {
075      LOG
076        .debug("Skipping recording bulk load entries in preCommitStoreFile for table:{}, family:{},"
077          + " Because the replication is not enabled", tableName, Bytes.toString(family));
078      return;
079    }
080
081    // This is completely cheating AND getting a HRegionServer from a RegionServerEnvironment is
082    // just going to break. This is all private. Not allowed. Regions shouldn't assume they are
083    // hosted in a RegionServer. TODO: fix.
084    RegionServerServices rss = ((HasRegionServerServices) env).getRegionServerServices();
085    Replication rep = (Replication) ((HRegionServer) rss).getReplicationSourceService();
086
087    rep.addHFileRefsToQueue(tableName, family, pairs);
088  }
089}