add threshold for refresh table entry with location

Author: maochongxin

src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java

@@ -401,6 +401,7 @@ private void initProperties() {
             routeMap.put(TABLE_ENTRY_ACQUIRE_CONNECT_TIMEOUT.getKey(), String.valueOf(tableEntryAcquireConnectTimeout));
             routeMap.put(TABLE_ENTRY_ACQUIRE_SOCKET_TIMEOUT.getKey(), String.valueOf(tableEntryAcquireSocketTimeout));
             routeMap.put(TABLE_ENTRY_REFRESH_INTERVAL_BASE.getKey(), String.valueOf(tableEntryRefreshIntervalBase));
+            routeMap.put(TABLE_ENTRY_LOCATION_REFRESH_THRESHOLD.getKey(), String.valueOf(tableEntryLocationRefreshThreshold));
             routeMap.put(TABLE_ENTRY_REFRESH_INTERVAL_CEILING.getKey(), String.valueOf(tableEntryRefreshIntervalCeiling));
             routeMap.put(TABLE_ENTRY_REFRESH_TRY_TIMES.getKey(), String.valueOf(tableEntryRefreshTryTimes));
         }

src/main/java/com/alipay/oceanbase/rpc/location/LocationUtil.java

@@ -47,6 +47,7 @@
 
 import static com.alipay.oceanbase.rpc.location.model.partition.ObPartitionKey.MAX_PARTITION_ELEMENT;
 import static com.alipay.oceanbase.rpc.location.model.partition.ObPartitionKey.MIN_PARTITION_ELEMENT;
+import static com.alipay.oceanbase.rpc.property.Property.TABLE_ENTRY_LOCATION_REFRESH_THRESHOLD;
 import static com.alipay.oceanbase.rpc.util.RandomUtil.getRandomNum;
 import static com.alipay.oceanbase.rpc.util.TableClientLoggerFactory.*;
 import static java.lang.String.format;
@@ -745,10 +746,14 @@ private static TableEntry getTableEntryFromRemote(Connection connection, TableEn
                 }
 
                 if (ObGlobal.obVsnMajor() >= 4) {
-                    // only set empty partitionEntry
-                    ObPartitionEntry partitionEntry = new ObPartitionEntry();
-                    tableEntry.setPartitionEntry(partitionEntry);
-                    tableEntry.setRefreshTimeMills(System.currentTimeMillis());
+                    // only set empty partitionEntry 
+                    if (tableEntry.getPartitionNum() <= TABLE_ENTRY_LOCATION_REFRESH_THRESHOLD.getDefaultLong()) {
+                        getTableEntryLocationFromRemote(connection, key, tableEntry);
+                    } else {
+                        ObPartitionEntry partitionEntry = new ObPartitionEntry();
+                        tableEntry.setPartitionEntry(partitionEntry);
+                        tableEntry.setRefreshTimeMills(System.currentTimeMillis());
+                    }
                 } else {
                     // get location info
                     getTableEntryLocationFromRemote(connection, key, tableEntry);
@@ -915,6 +920,7 @@ public static TableEntry getTableEntryLocationFromRemote(Connection connection,
                 ps.setString(1, key.getTenantName());
                 ps.setString(2, key.getDatabaseName());
                 ps.setString(3, key.getTableName());
+                ps.setString(4, key.getTenantName());
                 rs = ps.executeQuery();
                 partitionEntry = getPartitionLocationFromResultSet(tableEntry, rs, partitionEntry);
             } catch (Exception e) {
@@ -1279,28 +1285,55 @@ private static ObPartitionEntry getPartitionLocationFromResultSet(TableEntry tab
                 } else {
                     tabletLsIdMap.put(partitionId, INVALID_LS_ID); // non-partitioned table
                 }
+                ObPartitionLocationInfo partitionLocationInfo = partitionEntry
+                        .getPartitionInfo(partitionId);
+                ObPartitionLocation location = partitionLocationInfo.getPartitionLocation();
+                if (location == null) {
+                    partitionLocationInfo.rwLock.writeLock().lock();
+                    try {
+                        location = partitionLocationInfo.getPartitionLocation();
+                        if (location == null) {
+                            location = new ObPartitionLocation();
+                            partitionLocationInfo.updateLocation(location, lsId);
+                        }
+                    } finally {
+                        partitionLocationInfo.rwLock.writeLock().unlock();
+                    }
+                }
+                if (!replica.isValid()) {
+                    RUNTIME
+                            .warn(format(
+                                    "Replica is invalid; continuing. Replica=%s, PartitionId/TabletId=%d, TableId=%d",
+                                    replica, partitionId, tableEntry.getTableId()));
+                    continue;
+                }
+                location.addReplicaLocation(replica);
+
+                if (partitionLocationInfo.initialized.compareAndSet(false, true)) {
+                    partitionLocationInfo.initializationLatch.countDown();
+                }
             } else {
                 partitionId = rs.getLong("partition_id");
                 if (tableEntry.isPartitionTable()
-                    && null != tableEntry.getPartitionInfo().getSubPartDesc()) {
+                        && null != tableEntry.getPartitionInfo().getSubPartDesc()) {
                     partitionId = ObPartIdCalculator.getPartIdx(partitionId, tableEntry
-                        .getPartitionInfo().getSubPartDesc().getPartNum());
+                            .getPartitionInfo().getSubPartDesc().getPartNum());
                 }
-            }
-            if (!replica.isValid()) {
-                RUNTIME
-                    .warn(format(
-                        "replica is invalid, continue, replica=%s, partitionId/tabletId=%d, tableId=%d",
-                        replica, partitionId, tableEntry.getTableId()));
-                continue;
-            }
-            ObPartitionLocation location = partitionLocation.get(partitionId);
+                if (!replica.isValid()) {
+                    RUNTIME
+                            .warn(format(
+                                    "replica is invalid, continue, replica=%s, partitionId/tabletId=%d, tableId=%d",
+                                    replica, partitionId, tableEntry.getTableId()));
+                    continue;
+                }
+                ObPartitionLocation location = partitionLocation.get(partitionId);
 
-            if (location == null) {
-                location = new ObPartitionLocation();
-                partitionLocation.put(partitionId, location);
+                if (location == null) {
+                    location = new ObPartitionLocation();
+                    partitionLocation.put(partitionId, location);
+                }
+                location.addReplicaLocation(replica);
             }
-            location.addReplicaLocation(replica);
         }
 
         if (ObGlobal.obVsnMajor() < 4) {

src/main/java/com/alipay/oceanbase/rpc/property/Property.java

@@ -59,6 +59,8 @@ public enum Property {
     TABLE_ENTRY_ACQUIRE_SOCKET_TIMEOUT("table.entry.acquire.socket.timeout", 3000L,
                                        "刷新TABLE地址的SOCKET超时时间"),
 
+    TABLE_ENTRY_LOCATION_REFRESH_THRESHOLD("table.entry.location.refresh.threshold", 100L, "刷新TABLE ENTRY同时刷新LOCATION的阈值"),
+
     TABLE_ENTRY_REFRESH_INTERVAL_BASE("table.entry.refresh.interval.base", 100L, "刷新TABLE地址的基础时间间隔"),
 
     @Deprecated

src/main/java/com/alipay/oceanbase/rpc/table/AbstractObTableClient.java

@@ -50,6 +50,8 @@ public abstract class AbstractObTableClient extends AbstractTable {
     protected long            tableEntryRefreshIntervalBase             = TABLE_ENTRY_REFRESH_INTERVAL_BASE
                                                                             .getDefaultLong();
 
+    protected long            tableEntryLocationRefreshThreshold        = TABLE_ENTRY_LOCATION_REFRESH_THRESHOLD.getDefaultLong();
+    
     protected long            tableEntryRefreshIntervalCeiling          = TABLE_ENTRY_REFRESH_INTERVAL_CEILING
                                                                             .getDefaultLong();