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}