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;
019
020import static org.junit.Assert.assertArrayEquals;
021import static org.junit.Assert.assertEquals;
022import static org.junit.Assert.fail;
023
024import java.io.IOException;
025import java.util.ArrayList;
026import java.util.List;
027import java.util.Optional;
028import org.apache.hadoop.conf.Configuration;
029import org.apache.hadoop.hbase.ArrayBackedTag;
030import org.apache.hadoop.hbase.Cell;
031import org.apache.hadoop.hbase.CellUtil;
032import org.apache.hadoop.hbase.HBaseClassTestRule;
033import org.apache.hadoop.hbase.HBaseConfiguration;
034import org.apache.hadoop.hbase.HBaseTestingUtility;
035import org.apache.hadoop.hbase.HColumnDescriptor;
036import org.apache.hadoop.hbase.HConstants;
037import org.apache.hadoop.hbase.HTableDescriptor;
038import org.apache.hadoop.hbase.KeyValue;
039import org.apache.hadoop.hbase.KeyValueUtil;
040import org.apache.hadoop.hbase.PrivateCellUtil;
041import org.apache.hadoop.hbase.TableName;
042import org.apache.hadoop.hbase.Tag;
043import org.apache.hadoop.hbase.client.Admin;
044import org.apache.hadoop.hbase.client.Connection;
045import org.apache.hadoop.hbase.client.ConnectionFactory;
046import org.apache.hadoop.hbase.client.Durability;
047import org.apache.hadoop.hbase.client.Get;
048import org.apache.hadoop.hbase.client.Put;
049import org.apache.hadoop.hbase.client.Result;
050import org.apache.hadoop.hbase.client.Table;
051import org.apache.hadoop.hbase.client.replication.ReplicationAdmin;
052import org.apache.hadoop.hbase.codec.KeyValueCodecWithTags;
053import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
054import org.apache.hadoop.hbase.coprocessor.ObserverContext;
055import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
056import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
057import org.apache.hadoop.hbase.coprocessor.RegionObserver;
058import org.apache.hadoop.hbase.testclassification.LargeTests;
059import org.apache.hadoop.hbase.testclassification.ReplicationTests;
060import org.apache.hadoop.hbase.util.Bytes;
061import org.apache.hadoop.hbase.wal.WALEdit;
062import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster;
063import org.junit.AfterClass;
064import org.junit.BeforeClass;
065import org.junit.ClassRule;
066import org.junit.Test;
067import org.junit.experimental.categories.Category;
068import org.slf4j.Logger;
069import org.slf4j.LoggerFactory;
070
071@Category({ReplicationTests.class, LargeTests.class})
072public class TestReplicationWithTags {
073
074  @ClassRule
075  public static final HBaseClassTestRule CLASS_RULE =
076      HBaseClassTestRule.forClass(TestReplicationWithTags.class);
077
078  private static final Logger LOG = LoggerFactory.getLogger(TestReplicationWithTags.class);
079  private static final byte TAG_TYPE = 1;
080
081  private static Configuration conf1 = HBaseConfiguration.create();
082  private static Configuration conf2;
083
084  private static ReplicationAdmin replicationAdmin;
085
086  private static Connection connection1;
087  private static Connection connection2;
088
089  private static Table htable1;
090  private static Table htable2;
091
092  private static HBaseTestingUtility utility1;
093  private static HBaseTestingUtility utility2;
094  private static final long SLEEP_TIME = 500;
095  private static final int NB_RETRIES = 10;
096
097  private static final TableName TABLE_NAME = TableName.valueOf("TestReplicationWithTags");
098  private static final byte[] FAMILY = Bytes.toBytes("f");
099  private static final byte[] ROW = Bytes.toBytes("row");
100
101  @BeforeClass
102  public static void setUpBeforeClass() throws Exception {
103    conf1.setInt("hfile.format.version", 3);
104    conf1.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/1");
105    conf1.setInt("replication.source.size.capacity", 10240);
106    conf1.setLong("replication.source.sleepforretries", 100);
107    conf1.setInt("hbase.regionserver.maxlogs", 10);
108    conf1.setLong("hbase.master.logcleaner.ttl", 10);
109    conf1.setInt("zookeeper.recovery.retry", 1);
110    conf1.setInt("zookeeper.recovery.retry.intervalmill", 10);
111    conf1.setLong(HConstants.THREAD_WAKE_FREQUENCY, 100);
112    conf1.setInt("replication.stats.thread.period.seconds", 5);
113    conf1.setBoolean("hbase.tests.use.shortcircuit.reads", false);
114    conf1.setStrings(HConstants.REPLICATION_CODEC_CONF_KEY, KeyValueCodecWithTags.class.getName());
115    conf1.setStrings(CoprocessorHost.USER_REGION_COPROCESSOR_CONF_KEY,
116        TestCoprocessorForTagsAtSource.class.getName());
117
118    utility1 = new HBaseTestingUtility(conf1);
119    utility1.startMiniZKCluster();
120    MiniZooKeeperCluster miniZK = utility1.getZkCluster();
121    // Have to reget conf1 in case zk cluster location different
122    // than default
123    conf1 = utility1.getConfiguration();
124    replicationAdmin = new ReplicationAdmin(conf1);
125    LOG.info("Setup first Zk");
126
127    // Base conf2 on conf1 so it gets the right zk cluster.
128    conf2 = HBaseConfiguration.create(conf1);
129    conf2.setInt("hfile.format.version", 3);
130    conf2.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/2");
131    conf2.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 6);
132    conf2.setBoolean("hbase.tests.use.shortcircuit.reads", false);
133    conf2.setStrings(HConstants.REPLICATION_CODEC_CONF_KEY, KeyValueCodecWithTags.class.getName());
134    conf2.setStrings(CoprocessorHost.USER_REGION_COPROCESSOR_CONF_KEY,
135            TestCoprocessorForTagsAtSink.class.getName());
136
137    utility2 = new HBaseTestingUtility(conf2);
138    utility2.setZkCluster(miniZK);
139
140    LOG.info("Setup second Zk");
141    utility1.startMiniCluster(2);
142    utility2.startMiniCluster(2);
143
144    ReplicationPeerConfig rpc = new ReplicationPeerConfig();
145    rpc.setClusterKey(utility2.getClusterKey());
146    replicationAdmin.addPeer("2", rpc, null);
147
148    HTableDescriptor table = new HTableDescriptor(TABLE_NAME);
149    HColumnDescriptor fam = new HColumnDescriptor(FAMILY);
150    fam.setMaxVersions(3);
151    fam.setScope(HConstants.REPLICATION_SCOPE_GLOBAL);
152    table.addFamily(fam);
153    try (Connection conn = ConnectionFactory.createConnection(conf1);
154        Admin admin = conn.getAdmin()) {
155      admin.createTable(table, HBaseTestingUtility.KEYS_FOR_HBA_CREATE_TABLE);
156    }
157    try (Connection conn = ConnectionFactory.createConnection(conf2);
158        Admin admin = conn.getAdmin()) {
159      admin.createTable(table, HBaseTestingUtility.KEYS_FOR_HBA_CREATE_TABLE);
160    }
161    htable1 = utility1.getConnection().getTable(TABLE_NAME);
162    htable2 = utility2.getConnection().getTable(TABLE_NAME);
163  }
164
165  /**
166   * @throws java.lang.Exception
167   */
168  @AfterClass
169  public static void tearDownAfterClass() throws Exception {
170    utility2.shutdownMiniCluster();
171    utility1.shutdownMiniCluster();
172  }
173
174  @Test
175  public void testReplicationWithCellTags() throws Exception {
176    LOG.info("testSimplePutDelete");
177    Put put = new Put(ROW);
178    put.setAttribute("visibility", Bytes.toBytes("myTag3"));
179    put.addColumn(FAMILY, ROW, ROW);
180
181    htable1 = utility1.getConnection().getTable(TABLE_NAME);
182    htable1.put(put);
183
184    Get get = new Get(ROW);
185    try {
186      for (int i = 0; i < NB_RETRIES; i++) {
187        if (i == NB_RETRIES - 1) {
188          fail("Waited too much time for put replication");
189        }
190        Result res = htable2.get(get);
191        if (res.isEmpty()) {
192          LOG.info("Row not available");
193          Thread.sleep(SLEEP_TIME);
194        } else {
195          assertArrayEquals(ROW, res.value());
196          assertEquals(1, TestCoprocessorForTagsAtSink.tags.size());
197          Tag tag = TestCoprocessorForTagsAtSink.tags.get(0);
198          assertEquals(TAG_TYPE, tag.getType());
199          break;
200        }
201      }
202    } finally {
203      TestCoprocessorForTagsAtSink.tags = null;
204    }
205  }
206
207  public static class TestCoprocessorForTagsAtSource implements RegionCoprocessor, RegionObserver {
208    @Override
209    public Optional<RegionObserver> getRegionObserver() {
210      return Optional.of(this);
211    }
212
213    @Override
214    public void prePut(final ObserverContext<RegionCoprocessorEnvironment> e, final Put put,
215        final WALEdit edit, final Durability durability) throws IOException {
216      byte[] attribute = put.getAttribute("visibility");
217      byte[] cf = null;
218      List<Cell> updatedCells = new ArrayList<>();
219      if (attribute != null) {
220        for (List<? extends Cell> edits : put.getFamilyCellMap().values()) {
221          for (Cell cell : edits) {
222            KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
223            if (cf == null) {
224              cf = CellUtil.cloneFamily(kv);
225            }
226            Tag tag = new ArrayBackedTag(TAG_TYPE, attribute);
227            List<Tag> tagList = new ArrayList<>(1);
228            tagList.add(tag);
229
230            KeyValue newKV = new KeyValue(CellUtil.cloneRow(kv), 0, kv.getRowLength(),
231                CellUtil.cloneFamily(kv), 0, kv.getFamilyLength(), CellUtil.cloneQualifier(kv), 0,
232                kv.getQualifierLength(), kv.getTimestamp(),
233                KeyValue.Type.codeToType(kv.getTypeByte()), CellUtil.cloneValue(kv), 0,
234                kv.getValueLength(), tagList);
235            ((List<Cell>) updatedCells).add(newKV);
236          }
237        }
238        put.getFamilyCellMap().remove(cf);
239        // Update the family map
240        put.getFamilyCellMap().put(cf, updatedCells);
241      }
242    }
243  }
244
245  public static class TestCoprocessorForTagsAtSink implements RegionCoprocessor, RegionObserver {
246    public static List<Tag> tags = null;
247
248    @Override
249    public Optional<RegionObserver> getRegionObserver() {
250      return Optional.of(this);
251    }
252
253    @Override
254    public void postGetOp(ObserverContext<RegionCoprocessorEnvironment> e, Get get,
255        List<Cell> results) throws IOException {
256      if (results.size() > 0) {
257        // Check tag presence in the 1st cell in 1st Result
258        if (!results.isEmpty()) {
259          Cell cell = results.get(0);
260          tags = PrivateCellUtil.getTags(cell);
261        }
262      }
263    }
264  }
265}