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