Skip to content

Commit 4cf2cdc

Browse files
mymeiyimymeiyi
committed
HBASE-25929 RegionServer JVM crash when compaction
1 parent 63141bf commit 4cf2cdc

2 files changed

Lines changed: 173 additions & 18 deletions

File tree

‎hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java‎

Lines changed: 18 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -451,25 +451,25 @@ protected boolean performCompaction(FileDetails fd, InternalScanner scanner, Cel
451451
progress.cancel();
452452
return false;
453453
}
454-
if (kvs != null && bytesWrittenProgressForShippedCall > shippedCallSizeLimit) {
455-
if (lastCleanCell != null) {
456-
// HBASE-16931, set back sequence id to avoid affecting scan order unexpectedly.
457-
// ShipperListener will do a clone of the last cells it refer, so need to set back
458-
// sequence id before ShipperListener.beforeShipped
459-
PrivateCellUtil.setSequenceId(lastCleanCell, lastCleanCellSeqId);
460-
}
461-
// Clone the cells that are in the writer so that they are freed of references,
462-
// if they are holding any.
463-
((ShipperListener) writer).beforeShipped();
464-
// The SHARED block references, being read for compaction, will be kept in prevBlocks
465-
// list(See HFileScannerImpl#prevBlocks). In case of scan flow, after each set of cells
466-
// being returned to client, we will call shipped() which can clear this list. Here by
467-
// we are doing the similar thing. In between the compaction (after every N cells
468-
// written with collective size of 'shippedCallSizeLimit') we will call shipped which
469-
// may clear prevBlocks list.
470-
kvs.shipped();
471-
bytesWrittenProgressForShippedCall = 0;
454+
}
455+
if (kvs != null && bytesWrittenProgressForShippedCall > shippedCallSizeLimit) {
456+
if (lastCleanCell != null) {
457+
// HBASE-16931, set back sequence id to avoid affecting scan order unexpectedly.
458+
// ShipperListener will do a clone of the last cells it refer, so need to set back
459+
// sequence id before ShipperListener.beforeShipped
460+
PrivateCellUtil.setSequenceId(lastCleanCell, lastCleanCellSeqId);
472461
}
462+
// Clone the cells that are in the writer so that they are freed of references,
463+
// if they are holding any.
464+
((ShipperListener) writer).beforeShipped();
465+
// The SHARED block references, being read for compaction, will be kept in prevBlocks
466+
// list(See HFileScannerImpl#prevBlocks). In case of scan flow, after each set of cells
467+
// being returned to client, we will call shipped() which can clear this list. Here by
468+
// we are doing the similar thing. In between the compaction (after every N cells
469+
// written with collective size of 'shippedCallSizeLimit') we will call shipped which
470+
// may clear prevBlocks list.
471+
kvs.shipped();
472+
bytesWrittenProgressForShippedCall = 0;
473473
}
474474
if (lastCleanCell != null) {
475475
// HBASE-16931, set back sequence id to avoid affecting scan order unexpectedly

‎hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactionWithByteBuff.java‎

Lines changed: 155 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,155 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
package org.apache.hadoop.hbase.regionserver;
19+
20+
import java.io.IOException;
21+
import java.util.ArrayList;
22+
import java.util.List;
23+
24+
import org.apache.hadoop.conf.Configuration;
25+
import org.apache.hadoop.hbase.HBaseClassTestRule;
26+
import org.apache.hadoop.hbase.HBaseTestingUtility;
27+
import org.apache.hadoop.hbase.HConstants;
28+
import org.apache.hadoop.hbase.TableName;
29+
import org.apache.hadoop.hbase.client.Admin;
30+
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
31+
import org.apache.hadoop.hbase.client.Put;
32+
import org.apache.hadoop.hbase.client.Table;
33+
import org.apache.hadoop.hbase.client.TableDescriptor;
34+
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
35+
import org.apache.hadoop.hbase.io.ByteBuffAllocator;
36+
import org.apache.hadoop.hbase.testclassification.LargeTests;
37+
import org.apache.hadoop.hbase.util.Bytes;
38+
import org.apache.hadoop.hbase.util.JVMClusterUtil;
39+
import org.junit.AfterClass;
40+
import org.junit.BeforeClass;
41+
import org.junit.ClassRule;
42+
import org.junit.Rule;
43+
import org.junit.Test;
44+
import org.junit.experimental.categories.Category;
45+
import org.junit.rules.TestName;
46+
47+
@Category(LargeTests.class)
48+
public class TestCompactionWithByteBuff {
49+
@ClassRule
50+
public static final HBaseClassTestRule CLASS_RULE =
51+
HBaseClassTestRule.forClass(TestCompactionWithByteBuff.class);
52+
@Rule
53+
public TestName name = new TestName();
54+
55+
private static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
56+
private static Configuration conf = TEST_UTIL.getConfiguration();
57+
private static Admin admin = null;
58+
59+
private static final byte[] COLUMN = Bytes.toBytes("A");
60+
private static final int REGION_COUNT = 5;
61+
private static final long ROW_COUNT = 200;
62+
private static final int ROW_LENGTH = 20;
63+
private static final int VALUE_LENGTH = 5000;
64+
65+
@BeforeClass
66+
public static void setupBeforeClass() throws Exception {
67+
conf.setBoolean(ByteBuffAllocator.ALLOCATOR_POOL_ENABLED_KEY, true);
68+
conf.setInt(ByteBuffAllocator.BUFFER_SIZE_KEY, 1024 * 5);
69+
conf.setInt(CompactSplit.SMALL_COMPACTION_THREADS, REGION_COUNT * 2);
70+
conf.setInt(CompactSplit.LARGE_COMPACTION_THREADS, REGION_COUNT * 2);
71+
conf.set(HConstants.BUCKET_CACHE_IOENGINE_KEY, "offheap");
72+
conf.setInt(HConstants.BUCKET_CACHE_SIZE_KEY, 512);
73+
TEST_UTIL.startMiniCluster();
74+
admin = TEST_UTIL.getAdmin();
75+
}
76+
77+
@AfterClass
78+
public static void tearDownAfterClass() throws Exception {
79+
TEST_UTIL.shutdownMiniCluster();
80+
}
81+
82+
@Test
83+
public void testCompaction() throws Exception {
84+
TableName table = TableName.valueOf("t1");
85+
admin.compactionSwitch(false, new ArrayList<>(0));
86+
try (Table t = createTable(TEST_UTIL, table)) {
87+
for (int i = 0; i < 2; i++) {
88+
put(t);
89+
admin.flush(table);
90+
}
91+
admin.compactionSwitch(true, new ArrayList<>(0));
92+
admin.majorCompact(table);
93+
List<JVMClusterUtil.RegionServerThread> regionServerThreads =
94+
TEST_UTIL.getHBaseCluster().getRegionServerThreads();
95+
TEST_UTIL.waitFor(2 * 60 * 1000L, () -> {
96+
boolean result = true;
97+
for (JVMClusterUtil.RegionServerThread regionServerThread : regionServerThreads) {
98+
HRegionServer regionServer = regionServerThread.getRegionServer();
99+
List<HRegion> regions = regionServer.getRegions(table);
100+
for (HRegion region : regions) {
101+
List<String> storeFileList = region.getStoreFileList(new byte[][] { COLUMN });
102+
if (storeFileList.size() > 1) {
103+
result = false;
104+
}
105+
}
106+
}
107+
return result;
108+
});
109+
}
110+
}
111+
112+
private Table createTable(HBaseTestingUtility util, TableName tableName)
113+
throws IOException {
114+
TableDescriptor td =
115+
TableDescriptorBuilder.newBuilder(tableName)
116+
.setColumnFamily(
117+
ColumnFamilyDescriptorBuilder.newBuilder(COLUMN).setBlocksize(1024 * 4).build())
118+
.build();
119+
byte[][] splits = new byte[REGION_COUNT - 1][];
120+
for (int i = 1; i < REGION_COUNT; i++) {
121+
splits[i - 1] = Bytes.toBytes(buildRow((int) (ROW_COUNT / REGION_COUNT * i)));
122+
}
123+
return util.createTable(td, splits);
124+
}
125+
126+
private void put(Table table) throws IOException {
127+
for (int i = 0; i < ROW_COUNT; i++) {
128+
Put put = new Put(Bytes.toBytes(buildRow(i)));
129+
put.addColumn(COLUMN, Bytes.toBytes("filed01"), buildValue(i, 1));
130+
put.addColumn(COLUMN, Bytes.toBytes("filed02"), buildValue(i, 2));
131+
put.addColumn(COLUMN, Bytes.toBytes("filed03"), buildValue(i, 3));
132+
put.addColumn(COLUMN, Bytes.toBytes("filed04"), buildValue(i, 4));
133+
put.addColumn(COLUMN, Bytes.toBytes("filed05"), buildValue(i, 5));
134+
table.put(put);
135+
}
136+
}
137+
138+
private String buildRow(int index) {
139+
String value = Long.toString(index);
140+
String prefix = "user";
141+
for (int i = 0; i < ROW_LENGTH - value.length(); i++) {
142+
prefix += '0';
143+
}
144+
return prefix + value;
145+
}
146+
147+
private byte[] buildValue(int index, int qualifierId) {
148+
String row = buildRow(index) + "/f" + qualifierId + "-";
149+
StringBuffer result = new StringBuffer();
150+
while (result.length() < VALUE_LENGTH) {
151+
result.append(row);
152+
}
153+
return Bytes.toBytes(result.toString().substring(0, VALUE_LENGTH));
154+
}
155+
}

0 commit comments

Comments
 (0)