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.backup.util; 019 020import static org.apache.hadoop.hbase.wal.WALEdit.METAFAMILY; 021import static org.junit.jupiter.api.Assertions.assertEquals; 022import static org.junit.jupiter.api.Assertions.assertTrue; 023 024import java.io.IOException; 025import java.util.Arrays; 026import java.util.Collections; 027import java.util.List; 028import org.apache.hadoop.fs.Path; 029import org.apache.hadoop.hbase.Cell; 030import org.apache.hadoop.hbase.CellBuilderFactory; 031import org.apache.hadoop.hbase.CellBuilderType; 032import org.apache.hadoop.hbase.TableName; 033import org.apache.hadoop.hbase.testclassification.SmallTests; 034import org.apache.hadoop.hbase.util.Bytes; 035import org.apache.hadoop.hbase.wal.WAL; 036import org.apache.hadoop.hbase.wal.WALEdit; 037import org.apache.hadoop.hbase.wal.WALKeyImpl; 038import org.junit.jupiter.api.Tag; 039import org.junit.jupiter.api.Test; 040 041import org.apache.hbase.thirdparty.com.google.protobuf.ByteString; 042 043import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; 044import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos; 045 046/** 047 * Unit tests for {@link BulkLoadProcessor}. 048 * <p> 049 * These tests validate the extraction of bulk-loaded file paths from WAL entries under different 050 * scenarios, including: 051 * <ul> 052 * <li>Valid replicable bulk load entries</li> 053 * <li>Non-replicable bulk load entries</li> 054 * <li>Entries with no bulk load qualifier</li> 055 * <li>Entries containing multiple column families</li> 056 * </ul> 057 */ 058@Tag(SmallTests.TAG) 059public class TestBulkLoadProcessor { 060 061 /** 062 * Creates a WAL.Entry containing a {@link WALProtos.BulkLoadDescriptor} with the given 063 * parameters. 064 * @param tableName The table name 065 * @param regionName The encoded region name 066 * @param replicate Whether the bulk load is marked for replication 067 * @param family Column family name 068 * @param storeFiles One or more store file names to include 069 * @return A WAL.Entry representing the bulk load event 070 */ 071 private WAL.Entry createBulkLoadWalEntry(TableName tableName, String regionName, 072 boolean replicate, String family, String... storeFiles) { 073 074 // Build StoreDescriptor 075 WALProtos.StoreDescriptor.Builder storeDescBuilder = 076 WALProtos.StoreDescriptor.newBuilder().setFamilyName(ByteString.copyFromUtf8(family)) 077 .setStoreHomeDir(family).addAllStoreFile(Arrays.asList(storeFiles)); 078 079 // Build BulkLoadDescriptor 080 WALProtos.BulkLoadDescriptor.Builder bulkDescBuilder = WALProtos.BulkLoadDescriptor.newBuilder() 081 .setReplicate(replicate).setEncodedRegionName(ByteString.copyFromUtf8(regionName)) 082 .setTableName(ProtobufUtil.toProtoTableName(tableName)).setBulkloadSeqNum(1000) // Random 083 .addStores(storeDescBuilder); 084 085 byte[] value = bulkDescBuilder.build().toByteArray(); 086 087 // Build Cell with BULK_LOAD qualifier 088 Cell cell = CellBuilderFactory.create(CellBuilderType.DEEP_COPY).setType(Cell.Type.Put) 089 .setRow(new byte[] { 1 }).setFamily(METAFAMILY).setQualifier(WALEdit.BULK_LOAD) 090 .setValue(value).build(); 091 092 WALEdit edit = new WALEdit(); 093 edit.add(cell); 094 095 WALKeyImpl key = new WALKeyImpl(Bytes.toBytes(regionName), // region 096 tableName, 0L, 0L, null); 097 098 return new WAL.Entry(key, edit); 099 } 100 101 /** 102 * Verifies that a valid replicable bulk load WAL entry produces the correct number and structure 103 * of file paths. 104 */ 105 @Test 106 public void testProcessBulkLoadFiles_validEntry() throws IOException { 107 WAL.Entry entry = createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "region123", true, 108 "cf1", "file1", "file2"); 109 110 List<Path> paths = BulkLoadProcessor.processBulkLoadFiles(Collections.singletonList(entry)); 111 112 assertEquals(2, paths.size()); 113 assertTrue(paths.get(0).toString().contains("ns/tbl/region123/cf1/file1")); 114 assertTrue(paths.get(1).toString().contains("ns/tbl/region123/cf1/file2")); 115 } 116 117 @Test 118 public void testProcessBulkLoadFiles_validEntry_singleEntryApi() throws IOException { 119 WAL.Entry entry = createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "region123", true, 120 "cf1", "file1", "file2"); 121 122 List<Path> paths = BulkLoadProcessor.processBulkLoadFiles(entry.getKey(), entry.getEdit()); 123 124 assertEquals(2, paths.size()); 125 assertTrue(paths.get(0).toString().contains("ns/tbl/region123/cf1/file1")); 126 assertTrue(paths.get(1).toString().contains("ns/tbl/region123/cf1/file2")); 127 } 128 129 /** 130 * Verifies that a non-replicable bulk load entry is ignored. 131 */ 132 @Test 133 public void testProcessBulkLoadFiles_nonReplicableSkipped() throws IOException { 134 WAL.Entry entry = 135 createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "region123", false, "cf1", "file1"); 136 137 List<Path> paths = BulkLoadProcessor.processBulkLoadFiles(Collections.singletonList(entry)); 138 139 assertTrue(paths.isEmpty()); 140 } 141 142 @Test 143 public void testProcessBulkLoadFiles_nonReplicableSkipped_singleEntryApi() throws IOException { 144 WAL.Entry entry = 145 createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "region123", false, "cf1", "file1"); 146 147 List<Path> paths = BulkLoadProcessor.processBulkLoadFiles(entry.getKey(), entry.getEdit()); 148 149 assertTrue(paths.isEmpty()); 150 } 151 152 /** 153 * Verifies that entries without the BULK_LOAD qualifier are ignored. 154 */ 155 @Test 156 public void testProcessBulkLoadFiles_noBulkLoadQualifier() throws IOException { 157 WALEdit edit = new WALEdit(); 158 WALKeyImpl key = new WALKeyImpl(new byte[] {}, TableName.valueOf("ns", "tbl"), 0L, 0L, null); 159 WAL.Entry entry = new WAL.Entry(key, edit); 160 161 List<Path> paths = BulkLoadProcessor.processBulkLoadFiles(Collections.singletonList(entry)); 162 163 assertTrue(paths.isEmpty()); 164 } 165 166 @Test 167 public void testProcessBulkLoadFiles_noBulkLoadQualifier_singleEntryApi() throws IOException { 168 WALEdit edit = new WALEdit(); 169 WALKeyImpl key = new WALKeyImpl(new byte[] {}, TableName.valueOf("ns", "tbl"), 0L, 0L, null); 170 171 List<Path> paths = BulkLoadProcessor.processBulkLoadFiles(key, edit); 172 173 assertTrue(paths.isEmpty()); 174 } 175 176 /** 177 * Verifies that multiple WAL entries with different column families produce the correct set of 178 * file paths. 179 */ 180 @Test 181 public void testProcessBulkLoadFiles_multipleFamilies() throws IOException { 182 WAL.Entry entry = 183 createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "regionXYZ", true, "cf1", "file1"); 184 WAL.Entry entry2 = 185 createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "regionXYZ", true, "cf2", "fileA"); 186 187 List<Path> paths = BulkLoadProcessor.processBulkLoadFiles(Arrays.asList(entry, entry2)); 188 189 assertEquals(2, paths.size()); 190 assertTrue(paths.stream().anyMatch(p -> p.toString().contains("cf1/file1"))); 191 assertTrue(paths.stream().anyMatch(p -> p.toString().contains("cf2/fileA"))); 192 } 193 194 @Test 195 public void testProcessBulkLoadFiles_multipleFamilies_singleEntryApi() throws IOException { 196 WAL.Entry entry = 197 createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "regionXYZ", true, "cf1", "file1"); 198 WAL.Entry entry2 = 199 createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "regionXYZ", true, "cf2", "fileA"); 200 201 List<Path> paths1 = BulkLoadProcessor.processBulkLoadFiles(entry.getKey(), entry.getEdit()); 202 List<Path> paths2 = BulkLoadProcessor.processBulkLoadFiles(entry2.getKey(), entry2.getEdit()); 203 204 // combine to mimic processing multiple entries 205 paths1.addAll(paths2); 206 207 assertEquals(2, paths1.size()); 208 assertTrue(paths1.stream().anyMatch(p -> p.toString().contains("cf1/file1"))); 209 assertTrue(paths1.stream().anyMatch(p -> p.toString().contains("cf2/fileA"))); 210 } 211 212 /** 213 * Sanity check: list-based API should still work and return the same results as invoking the 214 * single-entry API for the same entry (ensures delegation/backwards compatibility). 215 */ 216 @Test 217 public void testProcessBulkLoadFiles_listApi_delegatesToSingle() throws IOException { 218 WAL.Entry entry = 219 createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "region123", true, "cf1", "file1"); 220 221 List<Path> single = BulkLoadProcessor.processBulkLoadFiles(entry.getKey(), entry.getEdit()); 222 List<Path> listApi = BulkLoadProcessor.processBulkLoadFiles(Collections.singletonList(entry)); 223 224 assertEquals(single.size(), listApi.size()); 225 assertTrue(listApi.get(0).toString().contains("ns/tbl/region123/cf1/file1")); 226 } 227}