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