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 static org.junit.Assert.assertEquals;
021import static org.junit.Assert.assertNull;
022import static org.mockito.Mockito.mock;
023import static org.mockito.Mockito.when;
024
025import java.util.ArrayList;
026import java.util.List;
027import java.util.NavigableMap;
028import java.util.OptionalLong;
029import org.apache.hadoop.conf.Configuration;
030import org.apache.hadoop.fs.FSDataInputStream;
031import org.apache.hadoop.fs.FSDataOutputStream;
032import org.apache.hadoop.fs.FileSystem;
033import org.apache.hadoop.fs.Path;
034import org.apache.hadoop.hbase.Cell;
035import org.apache.hadoop.hbase.CellBuilderFactory;
036import org.apache.hadoop.hbase.CellBuilderType;
037import org.apache.hadoop.hbase.HBaseClassTestRule;
038import org.apache.hadoop.hbase.HBaseTestingUtility;
039import org.apache.hadoop.hbase.HConstants;
040import org.apache.hadoop.hbase.ServerName;
041import org.apache.hadoop.hbase.TableName;
042import org.apache.hadoop.hbase.client.RegionInfo;
043import org.apache.hadoop.hbase.client.RegionInfoBuilder;
044import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl;
045import org.apache.hadoop.hbase.regionserver.wal.ProtobufLogWriter;
046import org.apache.hadoop.hbase.testclassification.MediumTests;
047import org.apache.hadoop.hbase.testclassification.ReplicationTests;
048import org.apache.hadoop.hbase.util.Bytes;
049import org.apache.hadoop.hbase.util.CommonFSUtils;
050import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
051import org.apache.hadoop.hbase.util.Pair;
052import org.apache.hadoop.hbase.wal.WAL;
053import org.apache.hadoop.hbase.wal.WALEdit;
054import org.apache.hadoop.hbase.wal.WALKeyImpl;
055import org.junit.AfterClass;
056import org.junit.BeforeClass;
057import org.junit.ClassRule;
058import org.junit.Test;
059import org.junit.experimental.categories.Category;
060
061import org.apache.hbase.thirdparty.com.google.common.io.ByteStreams;
062
063/**
064 * Enable compression and reset the WALEntryStream while reading in ReplicationSourceWALReader.
065 * <p/>
066 * This is used to confirm that we can work well when hitting EOFException in the middle when
067 * reading a WAL entry, when compression is enabled. See HBASE-27621 for more details.
068 */
069@Category({ ReplicationTests.class, MediumTests.class })
070public class TestWALEntryStreamCompressionReset {
071
072  @ClassRule
073  public static final HBaseClassTestRule CLASS_RULE =
074    HBaseClassTestRule.forClass(TestWALEntryStreamCompressionReset.class);
075
076  private static final HBaseTestingUtility UTIL = new HBaseTestingUtility();
077
078  private static TableName TABLE_NAME = TableName.valueOf("reset");
079
080  private static RegionInfo REGION_INFO = RegionInfoBuilder.newBuilder(TABLE_NAME).build();
081
082  private static byte[] FAMILY = Bytes.toBytes("family");
083
084  private static MultiVersionConcurrencyControl MVCC = new MultiVersionConcurrencyControl();
085
086  private static NavigableMap<byte[], Integer> SCOPE;
087
088  private static String GROUP_ID = "group";
089
090  private static FileSystem FS;
091
092  private static ReplicationSource SOURCE;
093
094  private static MetricsSource METRICS_SOURCE;
095
096  private static ReplicationSourceLogQueue LOG_QUEUE;
097
098  private static Path TEMPLATE_WAL_FILE;
099
100  private static int END_OFFSET_OF_WAL_ENTRIES;
101
102  private static Path WAL_FILE;
103
104  private static volatile long WAL_LENGTH;
105
106  private static ReplicationSourceWALReader READER;
107
108  // return the wal path, and also the end offset of last wal entry
109  private static Pair<Path, Long> generateWAL() throws Exception {
110    Path path = UTIL.getDataTestDir("wal");
111    ProtobufLogWriter writer = new ProtobufLogWriter();
112    writer.init(FS, path, UTIL.getConfiguration(), false, FS.getDefaultBlockSize(path), null);
113    for (int i = 0; i < Byte.MAX_VALUE; i++) {
114      WALEdit edit = new WALEdit();
115      edit.add(CellBuilderFactory.create(CellBuilderType.SHALLOW_COPY).setType(Cell.Type.Put)
116        .setRow(Bytes.toBytes(i)).setFamily(FAMILY).setQualifier(Bytes.toBytes("qualifier-" + i))
117        .setValue(Bytes.toBytes("v-" + i)).build());
118      writer.append(new WAL.Entry(new WALKeyImpl(REGION_INFO.getEncodedNameAsBytes(), TABLE_NAME,
119        EnvironmentEdgeManager.currentTime(), MVCC, SCOPE), edit));
120    }
121
122    WALEdit edit2 = new WALEdit();
123    edit2.add(CellBuilderFactory.create(CellBuilderType.SHALLOW_COPY).setType(Cell.Type.Put)
124      .setRow(Bytes.toBytes(-1)).setFamily(FAMILY).setQualifier(Bytes.toBytes("qualifier"))
125      .setValue(Bytes.toBytes("vv")).build());
126    edit2.add(CellBuilderFactory.create(CellBuilderType.SHALLOW_COPY).setType(Cell.Type.Put)
127      .setRow(Bytes.toBytes(-1)).setFamily(FAMILY).setQualifier(Bytes.toBytes("qualifier-1"))
128      .setValue(Bytes.toBytes("vvv")).build());
129    writer.append(new WAL.Entry(new WALKeyImpl(REGION_INFO.getEncodedNameAsBytes(), TABLE_NAME,
130      EnvironmentEdgeManager.currentTime(), MVCC, SCOPE), edit2));
131    writer.sync(false);
132    long offset = writer.getSyncedLength();
133    writer.close();
134    return Pair.newPair(path, offset);
135  }
136
137  @BeforeClass
138  public static void setUp() throws Exception {
139    Configuration conf = UTIL.getConfiguration();
140    FS = UTIL.getTestFileSystem();
141    conf.setBoolean(HConstants.ENABLE_WAL_COMPRESSION, true);
142    conf.setBoolean(CommonFSUtils.UNSAFE_STREAM_CAPABILITY_ENFORCE, false);
143    conf.setInt("replication.source.maxretriesmultiplier", 1);
144    FS.mkdirs(UTIL.getDataTestDir());
145    Pair<Path, Long> pair = generateWAL();
146    TEMPLATE_WAL_FILE = pair.getFirst();
147    END_OFFSET_OF_WAL_ENTRIES = pair.getSecond().intValue();
148    WAL_FILE = UTIL.getDataTestDir("rep_source");
149
150    METRICS_SOURCE = new MetricsSource("reset");
151    SOURCE = mock(ReplicationSource.class);
152    when(SOURCE.isPeerEnabled()).thenReturn(true);
153    when(SOURCE.getWALFileLengthProvider()).thenReturn(p -> OptionalLong.of(WAL_LENGTH));
154    when(SOURCE.getServerWALsBelongTo())
155      .thenReturn(ServerName.valueOf("localhost", 12345, EnvironmentEdgeManager.currentTime()));
156    when(SOURCE.getSourceMetrics()).thenReturn(METRICS_SOURCE);
157    ReplicationSourceManager rsm = new ReplicationSourceManager(null, null, conf, null, null, null,
158      null, null, null, mock(MetricsReplicationGlobalSourceSource.class));
159    when(SOURCE.getSourceManager()).thenReturn(rsm);
160
161    LOG_QUEUE = new ReplicationSourceLogQueue(conf, METRICS_SOURCE, SOURCE);
162    LOG_QUEUE.enqueueLog(WAL_FILE, GROUP_ID);
163    READER = new ReplicationSourceWALReader(FS, conf, LOG_QUEUE, 0, e -> e, SOURCE, GROUP_ID);
164  }
165
166  @AfterClass
167  public static void tearDown() throws Exception {
168    READER.setReaderRunning(false);
169    READER.join();
170    UTIL.cleanupTestDir();
171  }
172
173  private void test(byte[] content, FSDataOutputStream out) throws Exception {
174    // minus 15 so the second entry is incomplete
175    // 15 is a magic number here, we want the reader parse the first cell but not the second cell,
176    // especially not the qualifier of the second cell. The value of the second cell is 'vvv', which
177    // is 3 bytes, plus 8 bytes timestamp, and also qualifier, family and row(which should have been
178    // compressed), so 15 is a proper value, of course 14 or 16 could also work here.
179    out.write(content, 0, END_OFFSET_OF_WAL_ENTRIES - 15);
180    out.hflush();
181    WAL_LENGTH = END_OFFSET_OF_WAL_ENTRIES - 15;
182    READER.start();
183    List<WAL.Entry> entries = new ArrayList<>();
184    for (;;) {
185      WALEntryBatch batch = READER.poll(1000);
186      if (batch == null) {
187        break;
188      }
189      entries.addAll(batch.getWalEntries());
190    }
191    // should return all the entries except the last one
192    assertEquals(Byte.MAX_VALUE, entries.size());
193    for (int i = 0; i < Byte.MAX_VALUE; i++) {
194      WAL.Entry entry = entries.get(i);
195      assertEquals(1, entry.getEdit().size());
196      Cell cell = entry.getEdit().getCells().get(0);
197      assertEquals(i, Bytes.toInt(cell.getRowArray(), cell.getRowOffset()));
198      assertEquals(Bytes.toString(FAMILY),
199        Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));
200      assertEquals("qualifier-" + i, Bytes.toString(cell.getQualifierArray(),
201        cell.getQualifierOffset(), cell.getQualifierLength()));
202      assertEquals("v-" + i,
203        Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
204    }
205
206    // confirm that we can not get the last one since it is incomplete
207    assertNull(READER.poll(1000));
208    // write the last byte out
209    out.write(content, END_OFFSET_OF_WAL_ENTRIES - 15, 15);
210    out.hflush();
211    WAL_LENGTH = END_OFFSET_OF_WAL_ENTRIES;
212
213    // should get the last entry
214    WALEntryBatch batch = READER.poll(10000);
215    assertEquals(1, batch.getNbEntries());
216    WAL.Entry entry = batch.getWalEntries().get(0);
217    assertEquals(2, entry.getEdit().size());
218    Cell cell2 = entry.getEdit().getCells().get(0);
219    assertEquals(-1, Bytes.toInt(cell2.getRowArray(), cell2.getRowOffset()));
220    assertEquals(Bytes.toString(FAMILY),
221      Bytes.toString(cell2.getFamilyArray(), cell2.getFamilyOffset(), cell2.getFamilyLength()));
222    assertEquals("qualifier", Bytes.toString(cell2.getQualifierArray(), cell2.getQualifierOffset(),
223      cell2.getQualifierLength()));
224    assertEquals("vv",
225      Bytes.toString(cell2.getValueArray(), cell2.getValueOffset(), cell2.getValueLength()));
226
227    Cell cell3 = entry.getEdit().getCells().get(1);
228    assertEquals(-1, Bytes.toInt(cell3.getRowArray(), cell3.getRowOffset()));
229    assertEquals(Bytes.toString(FAMILY),
230      Bytes.toString(cell3.getFamilyArray(), cell3.getFamilyOffset(), cell3.getFamilyLength()));
231    assertEquals("qualifier-1", Bytes.toString(cell3.getQualifierArray(),
232      cell3.getQualifierOffset(), cell3.getQualifierLength()));
233    assertEquals("vvv",
234      Bytes.toString(cell3.getValueArray(), cell3.getValueOffset(), cell3.getValueLength()));
235  }
236
237  @Test
238  public void testReset() throws Exception {
239    byte[] content;
240    try (FSDataInputStream in = FS.open(TEMPLATE_WAL_FILE)) {
241      content = ByteStreams.toByteArray(in);
242    }
243    try (FSDataOutputStream out = FS.create(WAL_FILE)) {
244      test(content, out);
245    }
246  }
247}