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.coprocessor;
019
020import static org.junit.Assert.assertEquals;
021import static org.junit.Assert.assertTrue;
022import static org.junit.Assert.fail;
023
024import java.io.IOException;
025import java.util.List;
026import java.util.Optional;
027import java.util.stream.Collectors;
028
029import org.apache.hadoop.hbase.Cell;
030import org.apache.hadoop.hbase.CellBuilderType;
031import org.apache.hadoop.hbase.CellUtil;
032import org.apache.hadoop.hbase.ExtendedCellBuilderFactory;
033import org.apache.hadoop.hbase.HBaseClassTestRule;
034import org.apache.hadoop.hbase.HBaseTestingUtility;
035import org.apache.hadoop.hbase.TableName;
036import org.apache.hadoop.hbase.client.Append;
037import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
038import org.apache.hadoop.hbase.client.Connection;
039import org.apache.hadoop.hbase.client.Get;
040import org.apache.hadoop.hbase.client.Increment;
041import org.apache.hadoop.hbase.client.Mutation;
042import org.apache.hadoop.hbase.client.Result;
043import org.apache.hadoop.hbase.client.Table;
044import org.apache.hadoop.hbase.client.TableDescriptor;
045import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
046import org.apache.hadoop.hbase.client.TestFromClientSide;
047import org.apache.hadoop.hbase.regionserver.NoSuchColumnFamilyException;
048import org.apache.hadoop.hbase.testclassification.CoprocessorTests;
049import org.apache.hadoop.hbase.testclassification.MediumTests;
050import org.apache.hadoop.hbase.util.Bytes;
051import org.apache.hadoop.hbase.util.Pair;
052import org.junit.AfterClass;
053import org.junit.BeforeClass;
054import org.junit.ClassRule;
055import org.junit.Rule;
056import org.junit.Test;
057import org.junit.experimental.categories.Category;
058import org.junit.rules.TestName;
059import org.slf4j.Logger;
060import org.slf4j.LoggerFactory;
061
062/**
063 * Test coprocessor methods
064 * {@link RegionObserver#postIncrementBeforeWAL(ObserverContext, Mutation, List)} and
065 * {@link RegionObserver#postAppendBeforeWAL(ObserverContext, Mutation, List)}. These methods may
066 * change the cells which will be applied to memstore and WAL. So add unit test for the case which
067 * change the cell's column family.
068 */
069@Category({CoprocessorTests.class, MediumTests.class})
070public class TestPostIncrementAndAppendBeforeWAL {
071
072  @ClassRule
073  public static final HBaseClassTestRule CLASS_RULE =
074      HBaseClassTestRule.forClass(TestPostIncrementAndAppendBeforeWAL.class);
075
076  @Rule
077  public TestName name = new TestName();
078
079  private static final Logger LOG = LoggerFactory.getLogger(TestFromClientSide.class);
080
081  private static final HBaseTestingUtility UTIL = new HBaseTestingUtility();
082
083  private static Connection connection;
084
085  private static final byte [] ROW = Bytes.toBytes("row");
086  private static final String CF1 = "cf1";
087  private static final byte[] CF1_BYTES = Bytes.toBytes(CF1);
088  private static final String CF2 = "cf2";
089  private static final byte[] CF2_BYTES = Bytes.toBytes(CF2);
090  private static final String CF_NOT_EXIST = "cf_not_exist";
091  private static final byte[] CF_NOT_EXIST_BYTES = Bytes.toBytes(CF_NOT_EXIST);
092  private static final byte[] CQ1 = Bytes.toBytes("cq1");
093  private static final byte[] CQ2 = Bytes.toBytes("cq2");
094  private static final byte[] VALUE = Bytes.toBytes("value");
095
096  @BeforeClass
097  public static void setupBeforeClass() throws Exception {
098    UTIL.startMiniCluster();
099    connection = UTIL.getConnection();
100  }
101
102  @AfterClass
103  public static void tearDownAfterClass() throws Exception {
104    connection.close();
105    UTIL.shutdownMiniCluster();
106  }
107
108  private void createTableWithCoprocessor(TableName tableName, String coprocessor)
109      throws IOException {
110    TableDescriptor tableDesc = TableDescriptorBuilder.newBuilder(tableName)
111        .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(CF1_BYTES).build())
112        .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(CF2_BYTES).build())
113        .setCoprocessor(coprocessor).build();
114    connection.getAdmin().createTable(tableDesc);
115  }
116
117  @Test
118  public void testChangeCellWithDifferntColumnFamily() throws Exception {
119    TableName tableName = TableName.valueOf(name.getMethodName());
120    createTableWithCoprocessor(tableName,
121      ChangeCellWithDifferntColumnFamilyObserver.class.getName());
122
123    try (Table table = connection.getTable(tableName)) {
124      Increment increment = new Increment(ROW).addColumn(CF1_BYTES, CQ1, 1);
125      table.increment(increment);
126      Get get = new Get(ROW).addColumn(CF2_BYTES, CQ1);
127      Result result = table.get(get);
128      assertEquals(1, result.size());
129      assertEquals(1, Bytes.toLong(result.getValue(CF2_BYTES, CQ1)));
130
131      Append append = new Append(ROW).addColumn(CF1_BYTES, CQ2, VALUE);
132      table.append(append);
133      get = new Get(ROW).addColumn(CF2_BYTES, CQ2);
134      result = table.get(get);
135      assertEquals(1, result.size());
136      assertTrue(Bytes.equals(VALUE, result.getValue(CF2_BYTES, CQ2)));
137    }
138  }
139
140  @Test
141  public void testChangeCellWithNotExistColumnFamily() throws Exception {
142    TableName tableName = TableName.valueOf(name.getMethodName());
143    createTableWithCoprocessor(tableName,
144      ChangeCellWithNotExistColumnFamilyObserver.class.getName());
145
146    try (Table table = connection.getTable(tableName)) {
147      try {
148        Increment increment = new Increment(ROW).addColumn(CF1_BYTES, CQ1, 1);
149        table.increment(increment);
150        fail("should throw NoSuchColumnFamilyException");
151      } catch (Exception e) {
152        assertTrue(e instanceof NoSuchColumnFamilyException);
153      }
154      try {
155        Append append = new Append(ROW).addColumn(CF1_BYTES, CQ2, VALUE);
156        table.append(append);
157        fail("should throw NoSuchColumnFamilyException");
158      } catch (Exception e) {
159        assertTrue(e instanceof NoSuchColumnFamilyException);
160      }
161    }
162  }
163
164  public static class ChangeCellWithDifferntColumnFamilyObserver
165      implements RegionCoprocessor, RegionObserver {
166    @Override
167    public Optional<RegionObserver> getRegionObserver() {
168      return Optional.of(this);
169    }
170
171    @Override
172    public List<Pair<Cell, Cell>> postIncrementBeforeWAL(
173        ObserverContext<RegionCoprocessorEnvironment> ctx, Mutation mutation,
174        List<Pair<Cell, Cell>> cellPairs) throws IOException {
175      return cellPairs.stream()
176          .map(
177            pair -> new Pair<>(pair.getFirst(), newCellWithDifferentColumnFamily(pair.getSecond())))
178          .collect(Collectors.toList());
179    }
180
181    private Cell newCellWithDifferentColumnFamily(Cell cell) {
182      return ExtendedCellBuilderFactory.create(CellBuilderType.SHALLOW_COPY)
183          .setRow(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength())
184          .setFamily(CF2_BYTES, 0, CF2_BYTES.length).setQualifier(CellUtil.cloneQualifier(cell))
185          .setTimestamp(cell.getTimestamp()).setType(cell.getType().getCode())
186          .setValue(CellUtil.cloneValue(cell)).build();
187    }
188
189    @Override
190    public List<Pair<Cell, Cell>> postAppendBeforeWAL(
191        ObserverContext<RegionCoprocessorEnvironment> ctx, Mutation mutation,
192        List<Pair<Cell, Cell>> cellPairs) throws IOException {
193      return cellPairs.stream()
194          .map(
195            pair -> new Pair<>(pair.getFirst(), newCellWithDifferentColumnFamily(pair.getSecond())))
196          .collect(Collectors.toList());
197    }
198  }
199
200  public static class ChangeCellWithNotExistColumnFamilyObserver
201      implements RegionCoprocessor, RegionObserver {
202    @Override
203    public Optional<RegionObserver> getRegionObserver() {
204      return Optional.of(this);
205    }
206
207    @Override
208    public List<Pair<Cell, Cell>> postIncrementBeforeWAL(
209        ObserverContext<RegionCoprocessorEnvironment> ctx, Mutation mutation,
210        List<Pair<Cell, Cell>> cellPairs) throws IOException {
211      return cellPairs.stream()
212          .map(
213            pair -> new Pair<>(pair.getFirst(), newCellWithNotExistColumnFamily(pair.getSecond())))
214          .collect(Collectors.toList());
215    }
216
217    private Cell newCellWithNotExistColumnFamily(Cell cell) {
218      return ExtendedCellBuilderFactory.create(CellBuilderType.SHALLOW_COPY)
219          .setRow(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength())
220          .setFamily(CF_NOT_EXIST_BYTES, 0, CF_NOT_EXIST_BYTES.length)
221          .setQualifier(CellUtil.cloneQualifier(cell)).setTimestamp(cell.getTimestamp())
222          .setType(cell.getType().getCode()).setValue(CellUtil.cloneValue(cell)).build();
223    }
224
225    @Override
226    public List<Pair<Cell, Cell>> postAppendBeforeWAL(
227        ObserverContext<RegionCoprocessorEnvironment> ctx, Mutation mutation,
228        List<Pair<Cell, Cell>> cellPairs) throws IOException {
229      return cellPairs.stream()
230          .map(
231            pair -> new Pair<>(pair.getFirst(), newCellWithNotExistColumnFamily(pair.getSecond())))
232          .collect(Collectors.toList());
233    }
234  }
235}