Skip to content

Obkv support dds #390

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
maochongxin wants to merge 7 commits into
base: master
Choose a base branch
from
Open

Obkv support dds #390

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
ae3c550
replace Fastjson with Jackson (#370)
JackShi148 Aug 15, 2025
00a074f
fix IndexOutOfBoundsException when two same Get are in BatchOperation
WeiXinChan Jul 15, 2025
30b2728
Merge pull request #372 from oceanbase/fix_batch_out_of_index_1x
JackShi148 Aug 18, 2025
085a18f
fix get out of index
JackShi148 Aug 18, 2025
803ac8c
Merge pull request #373 from oceanbase/fix_batch_out_of_index_1x
JackShi148 Aug 18, 2025
2a2a040
Update pom.xml
wyfanxiao Aug 19, 2025
8c16a04
[maven-release-plugin] prepare release obkv-table-client-1.4.3
wyfanxiao Aug 19, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 11 additions & 6 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@

<groupId>com.oceanbase</groupId>
<artifactId>obkv-table-client</artifactId>
<version>1.4.2</version>
<version>1.4.3</version>

<name>${project.groupId}:${project.artifactId}</name>
<description>OceanBase JavaClient for TableApi</description>
Expand All @@ -27,7 +27,7 @@

<scm>
<connection>scm:git:https://github.com/oceanbase/obkv-table-client-java.git</connection>
<tag>obkv-table-client-1.4.2</tag>
<tag>obkv-table-client-1.4.3</tag>
<url>https://github.com/oceanbase/obkv-table-client-java</url>
</scm>

Expand Down Expand Up @@ -112,9 +112,9 @@
</dependency>

<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.83</version>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.19.0</version>
</dependency>

<dependency>
Expand Down Expand Up @@ -164,6 +164,10 @@
<artifactId>visible-assertions</artifactId>
<groupId>org.rnorth.visible-assertions</groupId>
</exclusion>
<exclusion>
<artifactId>jackson-annotations</artifactId>
<groupId>com.fasterxml.jackson.core</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
Expand Down Expand Up @@ -388,14 +392,15 @@
</plugin>
<plugin>
<artifactId>maven-shade-plugin</artifactId>
<version>3.2.4</version>
<version>3.6.0</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<createDependencyReducedPom>false</createDependencyReducedPom>
<shadedArtifactAttached>true</shadedArtifactAttached>
<shadedClassifierName>shade</shadedClassifierName>
<artifactSet>
Expand Down
35 changes: 24 additions & 11 deletions src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@

package com.alipay.oceanbase.rpc;

import com.alibaba.fastjson.JSON;
import com.alipay.oceanbase.rpc.checkandmutate.CheckAndInsUp;
import com.alipay.oceanbase.rpc.constant.Constants;
import com.alipay.oceanbase.rpc.exception.*;
Expand Down Expand Up @@ -46,6 +45,9 @@
import com.alipay.oceanbase.rpc.threadlocal.ThreadLocalMap;
import com.alipay.oceanbase.rpc.util.*;
import com.alipay.remoting.util.StringUtils;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import org.slf4j.Logger;

import java.util.*;
Expand All @@ -69,6 +71,15 @@
public class ObTableClient extends AbstractObTableClient implements Lifecycle {
private static final Logger logger = getLogger(ObTableClient.class);

private static final ObjectMapper objectMapper = new ObjectMapper();

static {
// FAIL_ON_EMPTY_BEANS means that whether throwing exception if there is no any serializable member with getter or setter in an object
// considering partitionElements in range partDesc is a list of Comparable interface and Comparable has no getter and setter
// we have to set this configuration as false because tableEntry may be serialized in debug log
objectMapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
}

private static final String usernameSeparators = ":;-;.";

private AtomicInteger tableEntryRefreshContinuousFailureCount = new AtomicInteger(
Expand Down Expand Up @@ -456,19 +467,19 @@ private void initMetadata() throws Exception {

List<ObServerAddr> rsList = ocpModel.getObServerAddrs();
BOOT.info("{} success to get rsList, paramURL: {}, rsList: {}，idc2Region: {}",
this.database, paramURL, JSON.toJSON(rsList), JSON.toJSON(ocpModel.getIdc2Region()));
this.database, paramURL, objectMapper.writeValueAsString(rsList), objectMapper.writeValueAsString(ocpModel.getIdc2Region()));

TableEntry tableEntry = loadTableEntryRandomly(rsList,//
rootServerKey,//
tableEntryAcquireConnectTimeout,//
tableEntryAcquireSocketTimeout, sysUA, initialized);
BOOT.info("{} success to get tableEntry with rootServerKey all_dummy_tables {}",
this.database, JSON.toJSON(tableEntry));
this.database, objectMapper.writeValueAsString(tableEntry));

List<ReplicaLocation> replicaLocations = tableEntry.getTableLocation()
.getReplicaLocations();
BOOT.info("{} success to get replicaLocation {}", this.database,
JSON.toJSON(replicaLocations));
objectMapper.writeValueAsString(replicaLocations));

for (ReplicaLocation replicaLocation : replicaLocations) {
ObServerInfo info = replicaLocation.getInfo();
Expand Down Expand Up @@ -500,11 +511,11 @@ private void initMetadata() throws Exception {
}
if (servers.isEmpty()) {
BOOT.error("{} failed to connect any replicaLocation server: {}", this.database,
JSON.toJSON(replicaLocations));
objectMapper.writeValueAsString(replicaLocations));
throw new Exception("failed to connect any replicaLocation server");
}

BOOT.info("{} success to build server connection {}", this.database, JSON.toJSON(servers));
BOOT.info("{} success to build server connection {}", this.database, objectMapper.writeValueAsString(servers));
this.tableRoster = tableRoster;
this.serverRoster.reset(servers);

Expand All @@ -525,7 +536,7 @@ private void initMetadata() throws Exception {

if (BOOT.isInfoEnabled()) {
BOOT.info("{} finish refresh serverRoster: {}", this.database,
JSON.toJSON(serverRoster));
objectMapper.writeValueAsString(serverRoster));
BOOT.info("finish initMetadata for all tables for database {}", this.database);
}

Expand Down Expand Up @@ -1031,7 +1042,7 @@ public void syncRefreshMetadata(boolean forceRenew) throws Exception {
currentIDC, regionFromOcp));

if (logger.isInfoEnabled()) {
logger.info("finish refresh serverRoster: {}", JSON.toJSON(serverRoster));
logger.info("finish refresh serverRoster: {}", objectMapper.writeValueAsString(serverRoster));
}
this.lastRefreshMetadataTimestamp = System.currentTimeMillis();
} finally {
Expand Down Expand Up @@ -1365,7 +1376,8 @@ public TableEntry getOrRefreshTableEntry(final String tableName, final boolean r
* @throws ObTableEntryRefreshException
*/
private TableEntry refreshTableEntry(TableEntry tableEntry, String tableName)
throws ObTableEntryRefreshException {
throws ObTableEntryRefreshException,
JsonProcessingException {
return refreshTableEntry(tableEntry, tableName, false);
}

Expand Down Expand Up @@ -1476,7 +1488,8 @@ public TableEntry refreshTableLocationByTabletId(TableEntry tableEntry, String t
* @throws ObTableEntryRefreshException
*/
private TableEntry refreshTableEntry(TableEntry tableEntry, String tableName, boolean fetchAll)
throws ObTableEntryRefreshException {
throws ObTableEntryRefreshException,
JsonProcessingException {
TableEntryKey tableEntryKey = new TableEntryKey(clusterName, tenantName, database,
tableName);
try {
Expand Down Expand Up @@ -1552,7 +1565,7 @@ private TableEntry refreshTableEntry(TableEntry tableEntry, String tableName, bo
if (logger.isDebugEnabled()) {
logger.debug(
"refresh table entry, dataSource: {}, tableName: {}, refresh: {} key:{} entry:{} ",
dataSourceName, tableName, true, tableEntryKey, JSON.toJSON(tableEntry));
dataSourceName, tableName, true, tableEntryKey, objectMapper.writeValueAsString(tableEntry));
}
return tableEntry;
}
Expand Down
7 changes: 4 additions & 3 deletions src/main/java/com/alipay/oceanbase/rpc/bolt/transport/ObTableConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@

package com.alipay.oceanbase.rpc.bolt.transport;

import com.alibaba.fastjson.JSONObject;
import com.alipay.oceanbase.rpc.ObGlobal;
import com.alipay.oceanbase.rpc.exception.*;
import com.alipay.oceanbase.rpc.location.LocationUtil;
Expand All @@ -26,6 +25,7 @@
import com.alipay.oceanbase.rpc.table.ObTable;
import com.alipay.oceanbase.rpc.util.*;
import com.alipay.remoting.Connection;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;

import java.net.ConnectException;
Expand All @@ -40,6 +40,7 @@ public class ObTableConnection {

private static final Logger LOGGER = TableClientLoggerFactory
.getLogger(ObTableConnection.class);
private static ObjectMapper objectMapper = new ObjectMapper();
private ObBytesString credential;
private long tenantId = 1; //默认值切勿不要随意改动
private Connection connection;
Expand Down Expand Up @@ -153,8 +154,8 @@ private void login() throws Exception {
// When the caller doesn't provide any parameters, configsMap is empty.
// In this case, we don't generate any JSON to avoid creating an empty object.
if (loginWithConfigs && !obTable.getConfigs().isEmpty()) {
JSONObject json = new JSONObject(obTable.getConfigs());
request.setConfigsStr(json.toJSONString());
String configStr = objectMapper.writeValueAsString(obTable.getConfigs());
request.setConfigsStr(configStr);
loginWithConfigs = false;
}
generatePassSecret(request);
Expand Down
37 changes: 21 additions & 16 deletions src/main/java/com/alipay/oceanbase/rpc/location/LocationUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,10 @@

package com.alipay.oceanbase.rpc.location;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.parser.ParserConfig;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.alipay.oceanbase.rpc.ObGlobal;
import com.alipay.oceanbase.rpc.constant.Constants;
import com.alipay.oceanbase.rpc.exception.*;
Expand Down Expand Up @@ -58,8 +58,13 @@ public class LocationUtil {

private static final Logger logger = TableClientLoggerFactory
.getLogger(LocationUtil.class);

private static final ObjectMapper objectMapper = new ObjectMapper();

static {
ParserConfig.getGlobalInstance().setSafeMode(true);
objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false)
.enable(DeserializationFeature.ACCEPT_EMPTY_STRING_AS_NULL_OBJECT);
loadJdbcDriver();
}

Expand Down Expand Up @@ -774,7 +779,7 @@ private static TableEntry getTableEntryFromRemote(Connection connection, TableEn

if (!initialized) {
if (BOOT.isInfoEnabled()) {
BOOT.info("get table entry from remote, entry={}", JSON.toJSON(tableEntry));
BOOT.info("get table entry from remote, entry={}", objectMapper.writeValueAsString(tableEntry));
}
} else {
if (logger.isInfoEnabled()) {
Expand All @@ -793,7 +798,7 @@ private static TableEntry getTableEntryFromRemote(Connection connection, TableEn
} catch (Exception e) {
RUNTIME.error(LCD.convert("01-00009"), key, e);
if (e instanceof ObTableEntryRefreshException) {
throw e;
throw (ObTableEntryRefreshException) e;
}
throw new ObTableEntryRefreshException(format(
"fail to get table entry from remote, key=%s", key), e);
Expand Down Expand Up @@ -1113,15 +1118,15 @@ private static void fetchFirstPart(Connection connection, TableEntry tableEntry,

if (logger.isInfoEnabled()) {
logger.info(format("uuid:%s, get first ranges from remote for %s, bounds=%s",
uuid, tableName, JSON.toJSON(bounds)));
uuid, tableName, objectMapper.writeValueAsString(bounds)));
}

} else if (obPartFuncType.isListPart()) {
Map<ObPartitionKey, Long> sets = parseFirstPartSets(rs, tableEntry);
((ObListPartDesc) tableEntry.getPartitionInfo().getFirstPartDesc()).setSets(sets);
if (logger.isInfoEnabled()) {
logger.info(format("uuid:%s, get first list sets from remote for %s, sets=%s",
uuid, tableName, JSON.toJSON(sets)));
uuid, tableName, objectMapper.writeValueAsString(sets)));
}
} else if (ObGlobal.obVsnMajor() >= 4
&& (obPartFuncType.isKeyPart() || obPartFuncType.isHashPart())) {
Expand Down Expand Up @@ -1186,14 +1191,14 @@ private static void fetchSubPart(Connection connection, TableEntry tableEntry,
.setHighBoundValues(highBoundVals);
if (logger.isInfoEnabled()) {
logger.info(format("uuid:%s, get sub ranges from remote for %s, bounds=%s",
uuid, tableName, JSON.toJSON(bounds)));
uuid, tableName, objectMapper.writeValueAsString(bounds)));
}
} else if (subPartFuncType.isListPart()) {
Map<ObPartitionKey, Long> sets = parseSubPartSets(rs, tableEntry);
((ObListPartDesc) tableEntry.getPartitionInfo().getSubPartDesc()).setSets(sets);
if (logger.isInfoEnabled()) {
logger.info(format("uuid:%s, get sub list sets from remote, sets=%s", uuid,
JSON.toJSON(sets)));
objectMapper.writeValueAsString(sets)));
}
} else if (ObGlobal.obVsnMajor() >= 4
&& (subPartFuncType.isKeyPart() || subPartFuncType.isHashPart())) {
Expand Down Expand Up @@ -1499,7 +1504,7 @@ private static void fetchPartitionInfo(Connection connection, TableEntry tableEn
info = parsePartitionInfo(rs);

if (logger.isInfoEnabled()) {
logger.info("get part info from remote info:{}", JSON.toJSON(info));
logger.info("get part info from remote info:{}", objectMapper.writeValueAsString(info));
}
tableEntry.setPartitionInfo(info);
} catch (SQLException e) {
Expand Down Expand Up @@ -2085,7 +2090,7 @@ private static OcpResponse getRemoteOcpResponseOrNull(String paramURL, String da
for (; tries < tryTimes; tries++) {
try {
content = loadStringFromUrl(paramURL, connectTimeout, readTimeout);
ocpResponse = JSONObject.parseObject(content, OcpResponse.class);
ocpResponse = objectMapper.readValue(content, OcpResponse.class);
if (ocpResponse != null && ocpResponse.validate()) {
if (dataSourceName != null && !dataSourceName.isEmpty()) {
saveLocalContent(dataSourceName, content);
Expand Down Expand Up @@ -2132,7 +2137,7 @@ private static OcpResponse getRemoteOcpIdcRegionOrNull(String paramURL, int conn
for (; tries < tryTimes; tries++) {
try {
content = loadStringFromUrl(paramURL, connectTimeout, readTimeout);
ocpResponse = JSONObject.parseObject(content, OcpResponse.class);
ocpResponse = objectMapper.readValue(content, OcpResponse.class);
if (ocpResponse != null) {
return ocpResponse;
}
Expand All @@ -2150,8 +2155,8 @@ private static OcpResponse getRemoteOcpIdcRegionOrNull(String paramURL, int conn
return null;
}

private static OcpResponse parseOcpResponse(String content) throws JSONException {
return JSONObject.parseObject(content, OcpResponse.class);
private static OcpResponse parseOcpResponse(String content) throws JsonProcessingException {
return objectMapper.readValue(content, OcpResponse.class);
}

private static OcpResponse getLocalOcpResponseOrNull(String fileName) {
Expand Down
10 changes: 10 additions & 0 deletions src/main/java/com/alipay/oceanbase/rpc/location/model/OcpResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@

package com.alipay.oceanbase.rpc.location.model;

import com.fasterxml.jackson.annotation.JsonProperty;

public class OcpResponse {
private int Code;
private String Message;
Expand All @@ -26,6 +28,7 @@ public class OcpResponse {
/*
* Get code.
*/
@JsonProperty("Code")
public int getCode() {
return Code;
}
Expand All @@ -40,6 +43,7 @@ public void setCode(int code) {
/*
* Get message.
*/
@JsonProperty("Message")
public String getMessage() {
return Message;
}
Expand All @@ -58,6 +62,11 @@ public boolean isSuccess() {
return Success;
}

@JsonProperty("Success")
public boolean getSuccess() {
return Success;
}

/*
* Set success.
*/
Expand All @@ -68,6 +77,7 @@ public void setSuccess(boolean success) {
/*
* Get data.
*/
@JsonProperty("Data")
public OcpResponseData getData() {
return Data;
}
Expand Down
Loading