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