Skip to content

Commit d4e53e4

Browse files
JackShi148JackShi148
authored
Merge pull request #154 from oceanbase/hbase_put_compatible
Hbase 1.x Put compatible
2 parents 074d59f + ae01a2b commit d4e53e4

File tree

15 files changed

+216
-192
lines changed

15 files changed

+216
-192
lines changed

‎src/main/java/com/alipay/oceanbase/hbase/OHTable.java‎

Lines changed: 88 additions & 131 deletions
Large diffs are not rendered by default.

‎src/main/java/com/alipay/oceanbase/hbase/OHTablePool.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import com.google.protobuf.Service;
2727
import com.google.protobuf.ServiceException;
2828
import org.apache.hadoop.conf.Configuration;
29+
import org.apache.hadoop.hbase.HBaseConfiguration;
2930
import org.apache.hadoop.hbase.HTableDescriptor;
3031
import org.apache.hadoop.hbase.TableName;
3132
import org.apache.hadoop.hbase.client.*;
@@ -67,7 +68,7 @@ public class OHTablePool implements Closeable {
6768
* Default Constructor. Default HBaseConfiguration and no limit on pool size.
6869
*/
6970
public OHTablePool() {
70-
this(new Configuration(), Integer.MAX_VALUE);
71+
this(HBaseConfiguration.create(), Integer.MAX_VALUE);
7172
}
7273

7374
/**

‎src/main/java/com/alipay/oceanbase/hbase/constants/OHConstants.java‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -143,10 +143,6 @@ public final class OHConstants {
143143

144144
public static final String MAX_KEYVALUE_SIZE_KEY = "hbase.client.keyvalue.maxsize";
145145

146-
public static final String HBASE_HTABLE_PUT_WRITE_BUFFER_CHECK = "hbase.htable.put.write.buffer.check";
147-
148-
public static final int DEFAULT_HBASE_HTABLE_PUT_WRITE_BUFFER_CHECK = 10;
149-
150146
public static final long WRITE_BUFFER_SIZE_DEFAULT = 2097152L;
151147

152148
public static final int MAX_KEYVALUE_SIZE_DEFAULT = -1;

‎src/main/java/com/alipay/oceanbase/hbase/util/BatchError.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,20 @@
1+
/*-
2+
* #%L
3+
* com.oceanbase:obkv-hbase-client
4+
* %%
5+
* Copyright (C) 2022 - 2025 OceanBase Group
6+
* %%
7+
* OBKV HBase Client Framework is licensed under Mulan PSL v2.
8+
* You can use this software according to the terms and conditions of the Mulan PSL v2.
9+
* You may obtain a copy of Mulan PSL v2 at:
10+
* http://license.coscl.org.cn/MulanPSL2
11+
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
12+
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
13+
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
14+
* See the Mulan PSL v2 for more details.
15+
* #L%
16+
*/
17+
118
package com.alipay.oceanbase.hbase.util;
219

320
import org.apache.hadoop.hbase.ServerName;

‎src/main/java/com/alipay/oceanbase/hbase/util/OHBufferedMutatorImpl.java‎

Lines changed: 47 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -18,9 +18,10 @@
1818
package com.alipay.oceanbase.hbase.util;
1919

2020
import com.alipay.oceanbase.hbase.OHTable;
21-
import com.google.common.annotations.VisibleForTesting;
21+
import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableBatchOperation;
2222
import org.apache.hadoop.classification.InterfaceAudience;
2323
import org.apache.hadoop.conf.Configuration;
24+
import org.apache.hadoop.hbase.KeyValue;
2425
import org.apache.hadoop.hbase.TableName;
2526
import org.apache.hadoop.hbase.client.*;
2627
import org.slf4j.Logger;
@@ -31,7 +32,9 @@
3132
import java.util.concurrent.ExecutorService;
3233
import java.util.concurrent.TimeUnit;
3334
import java.util.concurrent.atomic.AtomicLong;
34-
import static com.alipay.oceanbase.rpc.util.TableClientLoggerFactory.LCD;
35+
36+
import static com.alipay.oceanbase.hbase.util.TableHBaseLoggerFactory.LCD;
37+
import static com.alipay.oceanbase.rpc.ObGlobal.*;
3538

3639
@InterfaceAudience.Private
3740
public class OHBufferedMutatorImpl implements BufferedMutator {
@@ -40,13 +43,11 @@ public class OHBufferedMutatorImpl implements BufferedMutator {
4043

4144
private final ExceptionListener listener;
4245

43-
private final OHTable ohTable;
4446
private final TableName tableName;
4547
private volatile Configuration conf;
4648

47-
@VisibleForTesting
49+
private OHTable ohTable;
4850
final ConcurrentLinkedQueue<Mutation> asyncWriteBuffer = new ConcurrentLinkedQueue<Mutation>();
49-
@VisibleForTesting
5051
AtomicLong currentAsyncBufferSize = new AtomicLong(0);
5152

5253
private long writeBufferSize;
@@ -55,9 +56,11 @@ public class OHBufferedMutatorImpl implements BufferedMutator {
5556
private final ExecutorService pool;
5657
private int rpcTimeout;
5758
private int operationTimeout;
59+
private static final long OB_VERSION_4_2_5_1 = calcVersion(4, (short) 2,
60+
(byte) 5, (byte) 1);
5861

59-
public OHBufferedMutatorImpl(OHConnectionImpl ohConnection, BufferedMutatorParams params)
60-
throws IOException {
62+
public OHBufferedMutatorImpl(OHConnectionImpl ohConnection, BufferedMutatorParams params,
63+
OHTable ohTable) throws IOException {
6164
if (ohConnection == null || ohConnection.isClosed()) {
6265
throw new IllegalArgumentException("Connection is null or closed.");
6366
}
@@ -77,7 +80,11 @@ public OHBufferedMutatorImpl(OHConnectionImpl ohConnection, BufferedMutatorParam
7780
.getMaxKeyValueSize() : connectionConfig.getMaxKeyValueSize();
7881

7982
// create an OHTable object to do batch work
80-
this.ohTable = new OHTable(tableName, ohConnection, connectionConfig, pool);
83+
if (ohTable != null) {
84+
this.ohTable = ohTable;
85+
} else {
86+
this.ohTable = new OHTable(tableName, ohConnection, connectionConfig, pool);
87+
}
8188
}
8289

8390
@Override
@@ -119,14 +126,12 @@ public void mutate(List<? extends Mutation> mutations) throws IOException {
119126
validateOperation(m);
120127
toAddSize += m.heapSize();
121128
}
122-
123129
currentAsyncBufferSize.addAndGet(toAddSize);
124130
asyncWriteBuffer.addAll(mutations);
125131

126132
if (currentAsyncBufferSize.get() > writeBufferSize) {
127-
execute(false);
133+
batchExecute(false);
128134
}
129-
130135
}
131136

132137
/**
@@ -142,10 +147,18 @@ private void validateOperation(Mutation mt) throws IllegalArgumentException {
142147
}
143148
if (mt instanceof Put) {
144149
// family empty check is in validatePut
145-
HTable.validatePut((Put) mt, maxKeyValueSize);
146-
OHTable.checkFamilyViolation(mt.getFamilyMap().keySet(), true);
150+
OHTable.validatePut((Put) mt, maxKeyValueSize);
151+
if (isMultiFamilySupport()) {
152+
OHTable.checkFamilyViolation(mt.getFamilyMap().keySet(), true);
153+
} else {
154+
OHTable.checkFamilyViolationForOneFamily(mt.getFamilyMap().keySet());
155+
}
147156
} else {
148-
OHTable.checkFamilyViolation(mt.getFamilyMap().keySet(), false);
157+
if (isMultiFamilySupport()) {
158+
OHTable.checkFamilyViolation(mt.getFamilyMap().keySet(), false);
159+
} else {
160+
OHTable.checkFamilyViolationForOneFamily(mt.getFamilyMap().keySet());
161+
}
149162
}
150163
}
151164

@@ -156,7 +169,7 @@ private void validateOperation(Mutation mt) throws IllegalArgumentException {
156169
* @param flushAll - if true, sends all the writes and wait for all of them to finish before
157170
* returning.
158171
*/
159-
private void execute(boolean flushAll) throws IOException {
172+
private void batchExecute(boolean flushAll) throws IOException {
160173
LinkedList<Mutation> execBuffer = new LinkedList<>();
161174
long dequeuedSize = 0L;
162175
try {
@@ -172,19 +185,16 @@ private void execute(boolean flushAll) throws IOException {
172185
if (execBuffer.isEmpty()) {
173186
return;
174187
}
175-
ohTable.batch(execBuffer);
188+
Object[] results = new Object[execBuffer.size()];
189+
ohTable.batch(execBuffer, results);
176190
// if commit all successfully, clean execBuffer
177191
execBuffer.clear();
178192
} catch (Exception ex) {
179-
LOGGER.error(LCD.convert("01-00026"), ex);
193+
// do not recollect error operations, notify outside
194+
LOGGER.error("error happens: table name = ", tableName.getNameAsString(), ex);
180195
if (ex.getCause() instanceof RetriesExhaustedWithDetailsException) {
181-
LOGGER.error(tableName + ": One or more of the operations have failed after retries.");
196+
LOGGER.error(tableName.getNameAsString() + ": One or more of the operations have failed after retries.", ex);
182197
RetriesExhaustedWithDetailsException retryException = (RetriesExhaustedWithDetailsException) ex.getCause();
183-
// recollect mutations
184-
execBuffer.clear();
185-
for (int i = 0; i < retryException.getNumExceptions(); ++i) {
186-
execBuffer.add((Mutation) retryException.getRow(i));
187-
}
188198
if (listener != null) {
189199
listener.onException(retryException, this);
190200
} else {
@@ -194,12 +204,6 @@ private void execute(boolean flushAll) throws IOException {
194204
LOGGER.error("Errors unrelated to operations occur during mutation operation", ex);
195205
throw ex;
196206
}
197-
} finally {
198-
for (Mutation mutation : execBuffer) {
199-
long size = mutation.heapSize();
200-
currentAsyncBufferSize.addAndGet(size);
201-
asyncWriteBuffer.add(mutation);
202-
}
203207
}
204208
}
205209

@@ -209,7 +213,7 @@ public void close() throws IOException {
209213
return;
210214
}
211215
try {
212-
execute(true);
216+
batchExecute(true);
213217
} finally {
214218
// the pool in ObTableClient will be shut down too
215219
this.pool.shutdown();
@@ -234,13 +238,21 @@ public void setWriteBufferSize(long writeBufferSize) throws IOException {
234238
}
235239
}
236240

241+
/**
242+
* Only 4_2_5 BP1 - 4_3_0 and after 4_3_4 support multi-cf
243+
* */
244+
boolean isMultiFamilySupport() {
245+
return (OB_VERSION >= OB_VERSION_4_2_5_1 && OB_VERSION < OB_VERSION_4_3_0_0)
246+
|| (OB_VERSION >= OB_VERSION_4_3_4_0);
247+
}
248+
237249
/**
238250
* Force to commit all operations
239251
* do not care whether the pool is shut down or this BufferedMutator is closed
240252
*/
241253
@Override
242254
public void flush() throws IOException {
243-
execute(true);
255+
batchExecute(true);
244256
}
245257

246258
@Override
@@ -258,6 +270,10 @@ public void setOperationTimeout(int operationTimeout) {
258270
this.ohTable.setOperationTimeout(operationTimeout);
259271
}
260272

273+
public long getCurrentBufferSize() {
274+
return currentAsyncBufferSize.get();
275+
}
276+
261277
@Deprecated
262278
public List<Row> getWriteBuffer() {
263279
return Arrays.asList(asyncWriteBuffer.toArray(new Row[0]));

‎src/main/java/com/alipay/oceanbase/hbase/util/OHConnectionImpl.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -126,6 +126,11 @@ public BufferedMutator getBufferedMutator(TableName tableName) throws IOExceptio
126126

127127
@Override
128128
public BufferedMutator getBufferedMutator(BufferedMutatorParams params) throws IOException {
129+
return getBufferedMutator(params, null);
130+
}
131+
132+
public BufferedMutator getBufferedMutator(BufferedMutatorParams params, OHTable ohTable)
133+
throws IOException {
129134
if (params.getTableName() == null) {
130135
throw new IllegalArgumentException("TableName cannot be null.");
131136
}
@@ -138,7 +143,7 @@ public BufferedMutator getBufferedMutator(BufferedMutatorParams params) throws I
138143
if (params.getMaxKeyValueSize() == BUFFERED_PARAM_UNSET) {
139144
params.maxKeyValueSize(connectionConfig.getMaxKeyValueSize());
140145
}
141-
return new OHBufferedMutatorImpl(this, params);
146+
return new OHBufferedMutatorImpl(this, params, ohTable);
142147
}
143148

144149
@Override

‎src/main/java/com/alipay/oceanbase/hbase/util/OHTableFactory.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import com.alipay.oceanbase.hbase.OHTablePool;
2222
import org.apache.hadoop.classification.InterfaceAudience;
2323
import org.apache.hadoop.conf.Configuration;
24+
import org.apache.hadoop.hbase.HBaseConfiguration;
2425
import org.apache.hadoop.hbase.client.HTableFactory;
2526
import org.apache.hadoop.hbase.client.HTableInterface;
2627
import org.apache.hadoop.hbase.util.Bytes;
@@ -170,7 +171,7 @@ private Configuration adjustConfiguration(Configuration configuration, String ta
170171
}
171172

172173
private Configuration copyConfiguration(Configuration origin) {
173-
Configuration copy = new Configuration();
174+
Configuration copy = HBaseConfiguration.create();
174175

175176
for (Map.Entry<String, String> entry : origin) {
176177
copy.set(entry.getKey(), entry.getValue());

‎src/main/java/com/alipay/oceanbase/hbase/util/ObTableClientManager.java‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,6 @@
2121
import com.alipay.oceanbase.rpc.constant.Constants;
2222
import com.google.common.base.Objects;
2323
import org.apache.hadoop.classification.InterfaceAudience;
24-
import org.apache.hadoop.conf.Configuration;
25-
import org.apache.hadoop.hbase.client.ConnectionConfiguration;
2624

2725
import java.io.IOException;
2826
import java.util.Map;

‎src/test/java/com/alipay/oceanbase/hbase/HTableTestBase.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5189,7 +5189,7 @@ public void testCellTTL() throws Exception {
51895189
try {
51905190
tryPut(hTable, errorPut);
51915191
} catch (Exception e) {
5192-
assertTrue(e.getCause().toString().contains("Unknown column 'TTL'"));
5192+
assertTrue(e.getCause().getCause().toString().contains("Unknown column 'TTL'"));
51935193
}
51945194
// test put and get
51955195
tryPut(hTable, put1);

‎src/test/java/com/alipay/oceanbase/hbase/NativeHBaseTest.java‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
import org.apache.hadoop.hbase.TableName;
55
import org.apache.hadoop.hbase.client.*;
66
import org.junit.AfterClass;
7+
import org.junit.Assert;
78
import org.junit.Before;
89
import org.junit.BeforeClass;
910

@@ -39,8 +40,13 @@ public void cleanData() throws IOException {
3940

4041
@AfterClass
4142
public static void finish() throws IOException {
42-
hTable.close();
43-
multiCfHTable.close();
43+
try {
44+
hTable.close();
45+
multiCfHTable.close();
46+
} catch (Exception e) {
47+
Assert.assertSame(e.getClass(), IOException.class);
48+
Assert.assertTrue(e.getMessage().contains("put table"));
49+
}
4450
}
4551

4652
}

0 commit comments

Comments
 (0)