You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by GitBox <gi...@apache.org> on 2022/07/20 09:53:53 UTC
[GitHub] [iotdb] RYH61 opened a new pull request, #6735: [IOTDB-3880] Optimize asyncclientpool retry logic
RYH61 opened a new pull request, #6735:
URL: https://github.com/apache/iotdb/pull/6735
Optimize asyncclientpool retry logic
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] Beyyes commented on a diff in pull request #6735: [IOTDB-3880] Optimize asyncclientpool retry logic
Posted by GitBox <gi...@apache.org>.
Beyyes commented on code in PR #6735:
URL: https://github.com/apache/iotdb/pull/6735#discussion_r926195691
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/CreateRegionHandler.java:
##########
@@ -38,23 +38,26 @@ public class CreateRegionHandler extends AbstractRetryHandler
private static final Logger LOGGER = LoggerFactory.getLogger(CreateRegionHandler.class);
+ private int index;
// Used for Logger
Review Comment:
add a blank
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/AbstractRetryHandler.java:
##########
@@ -22,19 +22,15 @@
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.confignode.client.DataNodeRequestType;
-import java.util.Map;
+import java.util.List;
import java.util.concurrent.CountDownLatch;
public abstract class AbstractRetryHandler {
- protected final int index;
-
protected CountDownLatch countDownLatch;
- /**
- * Map<Index, TDataNodeLocation> The DataNode that successfully execute the request will be
- * removed from this map
- */
- protected Map<Integer, TDataNodeLocation> dataNodeLocations;
+
+ /** The DataNode that successfully execute the request will be removed from this list */
+ protected List<TDataNodeLocation> dataNodeLocations;
protected DataNodeRequestType dataNodeRequestType;
/** Target DataNode */
Review Comment:
add a blank
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/CreateRegionHandler.java:
##########
@@ -38,23 +38,26 @@ public class CreateRegionHandler extends AbstractRetryHandler
private static final Logger LOGGER = LoggerFactory.getLogger(CreateRegionHandler.class);
+ private int index;
// Used for Logger
private final TConsensusGroupId consensusGroupId;
public CreateRegionHandler(
- int index,
CountDownLatch latch,
+ DataNodeRequestType requestType,
TConsensusGroupId consensusGroupId,
TDataNodeLocation targetDataNode,
- Map<Integer, TDataNodeLocation> dataNodeLocations) {
- super(latch, DataNodeRequestType.CREATE_REGIONS, targetDataNode, dataNodeLocations, index);
+ List<TDataNodeLocation> dataNodeLocations,
+ int index) {
+ super(latch, requestType, targetDataNode, dataNodeLocations);
+ this.index = index;
this.consensusGroupId = consensusGroupId;
}
@Override
public void onComplete(TSStatus tsStatus) {
if (tsStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- getDataNodeLocations().remove(index);
+ dataNodeLocations.remove(index);
Review Comment:
why using `remove(index)` not `remove(targetDataNode)`.
`remove(index)` will change the original order of the list, using this operation must be cautious
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java:
##########
@@ -168,60 +190,86 @@ public void sendAsyncRequestToDataNode(
public void createRegions(
CreateRegionGroupsPlan createRegionGroupsPlan, Map<String, Long> ttlMap) {
- // Number of regions to be created
- int regionNum = 0;
- // Assign an independent index to each Region
+ List<TDataNodeLocation> dataNodeLocations = new ArrayList<>();
+ // Count the datanodes to be sent
for (Map.Entry<String, List<TRegionReplicaSet>> entry :
createRegionGroupsPlan.getRegionGroupMap().entrySet()) {
for (TRegionReplicaSet regionReplicaSet : entry.getValue()) {
- regionNum += regionReplicaSet.getDataNodeLocationsSize();
+ regionReplicaSet
+ .getDataNodeLocations()
Review Comment:
can be replaced with `dataNodeLocations.addAll(regionReplicaSet.getDataNodeLocations());`
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java:
##########
@@ -168,60 +190,86 @@ public void sendAsyncRequestToDataNode(
public void createRegions(
CreateRegionGroupsPlan createRegionGroupsPlan, Map<String, Long> ttlMap) {
- // Number of regions to be created
- int regionNum = 0;
- // Assign an independent index to each Region
+ List<TDataNodeLocation> dataNodeLocations = new ArrayList<>();
+ // Count the datanodes to be sent
for (Map.Entry<String, List<TRegionReplicaSet>> entry :
createRegionGroupsPlan.getRegionGroupMap().entrySet()) {
for (TRegionReplicaSet regionReplicaSet : entry.getValue()) {
- regionNum += regionReplicaSet.getDataNodeLocationsSize();
+ regionReplicaSet
+ .getDataNodeLocations()
+ .forEach(dataNodeLocation -> dataNodeLocations.add(dataNodeLocation));
}
}
- Map<Integer, AbstractRetryHandler> handlerMap = new ConcurrentHashMap<>();
- ConcurrentHashMap<Integer, TDataNodeLocation> dataNodeLocations = new ConcurrentHashMap<>();
- Map<Integer, Object> req = new ConcurrentHashMap<>();
+ if (dataNodeLocations.isEmpty()) {
+ return;
+ }
AtomicInteger index = new AtomicInteger();
- CountDownLatch latch = new CountDownLatch(regionNum);
- createRegionGroupsPlan
- .getRegionGroupMap()
- .forEach(
- (storageGroup, regionReplicaSets) -> {
- // Enumerate each RegionReplicaSet
- regionReplicaSets.forEach(
- regionReplicaSet -> {
- // Enumerate each Region
- regionReplicaSet
- .getDataNodeLocations()
- .forEach(
- dataNodeLocation -> {
- handlerMap.put(
- index.get(),
- new CreateRegionHandler(
- index.get(),
- latch,
- regionReplicaSet.getRegionId(),
- dataNodeLocation,
- dataNodeLocations));
-
- switch (regionReplicaSet.getRegionId().getType()) {
- case SchemaRegion:
- req.put(
- index.get(),
- genCreateSchemaRegionReq(storageGroup, regionReplicaSet));
- break;
- case DataRegion:
- req.put(
- index.get(),
- genCreateDataRegionReq(
- storageGroup,
- regionReplicaSet,
- ttlMap.get(storageGroup)));
- }
- dataNodeLocations.put(index.getAndIncrement(), dataNodeLocation);
- });
- });
- });
- sendAsyncRequestToDataNodeWithRetry(req, handlerMap, dataNodeLocations);
+ CountDownLatch countDownLatch = new CountDownLatch(dataNodeLocations.size());
+ for (int retry = 0; retry < retryNum; retry++) {
+ AbstractRetryHandler handler = null;
+ for (Map.Entry<String, List<TRegionReplicaSet>> entry :
Review Comment:
we only the `entry.getValue()`
so the foreach of `createRegionGroupsPlan.getRegionGroupMap().entrySet()` can be replaced with `createRegionGroupsPlan.getRegionGroupMap().values()`
same as below
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java:
##########
@@ -72,31 +72,63 @@ private AsyncDataNodeClientPool() {
* receive the requests
*
* @param req request
- * @param handlerMap Map<index, Handler>
- * @param dataNodeLocations ConcurrentHashMap<index, TDataNodeLocation> The specific DataNodes
+ * @param dataNodeLocations The specific DataNodes
+ * @param requestType DataNodeRequestType
+ * @param dataNodeResponseStatus request response list
Review Comment:
`request response list`, may be `response list`?
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java:
##########
@@ -72,31 +72,63 @@ private AsyncDataNodeClientPool() {
* receive the requests
*
* @param req request
- * @param handlerMap Map<index, Handler>
- * @param dataNodeLocations ConcurrentHashMap<index, TDataNodeLocation> The specific DataNodes
+ * @param dataNodeLocations The specific DataNodes
+ * @param requestType DataNodeRequestType
+ * @param dataNodeResponseStatus request response list
*/
public void sendAsyncRequestToDataNodeWithRetry(
Object req,
- Map<Integer, AbstractRetryHandler> handlerMap,
- Map<Integer, TDataNodeLocation> dataNodeLocations) {
- CountDownLatch countDownLatch = new CountDownLatch(dataNodeLocations.size());
+ List<TDataNodeLocation> dataNodeLocations,
+ DataNodeRequestType requestType,
+ List<TSStatus> dataNodeResponseStatus) {
if (dataNodeLocations.isEmpty()) {
return;
}
+ CountDownLatch countDownLatch = new CountDownLatch(dataNodeLocations.size());
+ AbstractRetryHandler handler = null;
for (int retry = 0; retry < retryNum; retry++) {
- AbstractRetryHandler handler = null;
- for (Map.Entry<Integer, TDataNodeLocation> entry : dataNodeLocations.entrySet()) {
- handler = handlerMap.get(entry.getKey());
+ for (TDataNodeLocation targetDataNode : dataNodeLocations) {
+ switch (requestType) {
+ case SET_TTL:
+ handler =
+ new SetTTLHandler(countDownLatch, requestType, targetDataNode, dataNodeLocations);
+ break;
+ case CREATE_FUNCTION:
+ case DROP_FUNCTION:
Review Comment:
CREATE_FUNCTION, DROP_FUNCTION and FLUSH have the parameter `dataNodeResponseStatus`,
however SET_TTL and UPDATE_REGION_ROUTE_MAP don't need this parameter, we can explain more details about this parameter
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] wangchao316 merged pull request #6735: [IOTDB-3880] Optimize asyncclientpool retry logic
Posted by GitBox <gi...@apache.org>.
wangchao316 merged PR #6735:
URL: https://github.com/apache/iotdb/pull/6735
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] RYH61 commented on a diff in pull request #6735: [IOTDB-3880] Optimize asyncclientpool retry logic
Posted by GitBox <gi...@apache.org>.
RYH61 commented on code in PR #6735:
URL: https://github.com/apache/iotdb/pull/6735#discussion_r926233056
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/AbstractRetryHandler.java:
##########
@@ -22,19 +22,15 @@
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.confignode.client.DataNodeRequestType;
-import java.util.Map;
+import java.util.List;
import java.util.concurrent.CountDownLatch;
public abstract class AbstractRetryHandler {
- protected final int index;
-
protected CountDownLatch countDownLatch;
- /**
- * Map<Index, TDataNodeLocation> The DataNode that successfully execute the request will be
- * removed from this map
- */
- protected Map<Integer, TDataNodeLocation> dataNodeLocations;
+
+ /** The DataNode that successfully execute the request will be removed from this list */
+ protected List<TDataNodeLocation> dataNodeLocations;
protected DataNodeRequestType dataNodeRequestType;
/** Target DataNode */
Review Comment:
fixed~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #6735: [IOTDB-3880] Optimize asyncclientpool retry logic
Posted by GitBox <gi...@apache.org>.
CRZbulabula commented on code in PR #6735:
URL: https://github.com/apache/iotdb/pull/6735#discussion_r927275336
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/AbstractRetryHandler.java:
##########
@@ -27,15 +27,11 @@
public abstract class AbstractRetryHandler {
- protected final int index;
-
protected CountDownLatch countDownLatch;
- /**
- * Map<Index, TDataNodeLocation> The DataNode that successfully execute the request will be
- * removed from this map
- */
- protected Map<Integer, TDataNodeLocation> dataNodeLocations;
+ /** The DataNode that successfully execute the request will be removed from this list */
Review Comment:
```suggestion
/** Map<DataNodeId, TDataNodeLocation> The DataNode that successfully execute the request will be removed from this list */
```
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/CreateRegionHandler.java:
##########
@@ -38,23 +38,27 @@ public class CreateRegionHandler extends AbstractRetryHandler
private static final Logger LOGGER = LoggerFactory.getLogger(CreateRegionHandler.class);
+ // The index of dataNodeLocations
+ private final int index;
Review Comment:
Add more detailed annotation: We use Index instead of DataNodeId because it is possible to send multiple createRegion requests to the same DataNode
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] RYH61 commented on a diff in pull request #6735: [IOTDB-3880] Optimize asyncclientpool retry logic
Posted by GitBox <gi...@apache.org>.
RYH61 commented on code in PR #6735:
URL: https://github.com/apache/iotdb/pull/6735#discussion_r926233260
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java:
##########
@@ -168,60 +190,86 @@ public void sendAsyncRequestToDataNode(
public void createRegions(
CreateRegionGroupsPlan createRegionGroupsPlan, Map<String, Long> ttlMap) {
- // Number of regions to be created
- int regionNum = 0;
- // Assign an independent index to each Region
+ List<TDataNodeLocation> dataNodeLocations = new ArrayList<>();
+ // Count the datanodes to be sent
for (Map.Entry<String, List<TRegionReplicaSet>> entry :
createRegionGroupsPlan.getRegionGroupMap().entrySet()) {
for (TRegionReplicaSet regionReplicaSet : entry.getValue()) {
- regionNum += regionReplicaSet.getDataNodeLocationsSize();
+ regionReplicaSet
+ .getDataNodeLocations()
Review Comment:
fixed~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] RYH61 commented on pull request #6735: [IOTDB-3880] Optimize asyncclientpool retry logic
Posted by GitBox <gi...@apache.org>.
RYH61 commented on PR #6735:
URL: https://github.com/apache/iotdb/pull/6735#issuecomment-1191168494
>
>  Will these methods produced npe?
fixed~This part has been modified
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] RYH61 commented on a diff in pull request #6735: [IOTDB-3880] Optimize asyncclientpool retry logic
Posted by GitBox <gi...@apache.org>.
RYH61 commented on code in PR #6735:
URL: https://github.com/apache/iotdb/pull/6735#discussion_r926233730
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java:
##########
@@ -168,60 +190,86 @@ public void sendAsyncRequestToDataNode(
public void createRegions(
CreateRegionGroupsPlan createRegionGroupsPlan, Map<String, Long> ttlMap) {
- // Number of regions to be created
- int regionNum = 0;
- // Assign an independent index to each Region
+ List<TDataNodeLocation> dataNodeLocations = new ArrayList<>();
+ // Count the datanodes to be sent
for (Map.Entry<String, List<TRegionReplicaSet>> entry :
createRegionGroupsPlan.getRegionGroupMap().entrySet()) {
for (TRegionReplicaSet regionReplicaSet : entry.getValue()) {
- regionNum += regionReplicaSet.getDataNodeLocationsSize();
+ regionReplicaSet
+ .getDataNodeLocations()
+ .forEach(dataNodeLocation -> dataNodeLocations.add(dataNodeLocation));
}
}
- Map<Integer, AbstractRetryHandler> handlerMap = new ConcurrentHashMap<>();
- ConcurrentHashMap<Integer, TDataNodeLocation> dataNodeLocations = new ConcurrentHashMap<>();
- Map<Integer, Object> req = new ConcurrentHashMap<>();
+ if (dataNodeLocations.isEmpty()) {
+ return;
+ }
AtomicInteger index = new AtomicInteger();
- CountDownLatch latch = new CountDownLatch(regionNum);
- createRegionGroupsPlan
- .getRegionGroupMap()
- .forEach(
- (storageGroup, regionReplicaSets) -> {
- // Enumerate each RegionReplicaSet
- regionReplicaSets.forEach(
- regionReplicaSet -> {
- // Enumerate each Region
- regionReplicaSet
- .getDataNodeLocations()
- .forEach(
- dataNodeLocation -> {
- handlerMap.put(
- index.get(),
- new CreateRegionHandler(
- index.get(),
- latch,
- regionReplicaSet.getRegionId(),
- dataNodeLocation,
- dataNodeLocations));
-
- switch (regionReplicaSet.getRegionId().getType()) {
- case SchemaRegion:
- req.put(
- index.get(),
- genCreateSchemaRegionReq(storageGroup, regionReplicaSet));
- break;
- case DataRegion:
- req.put(
- index.get(),
- genCreateDataRegionReq(
- storageGroup,
- regionReplicaSet,
- ttlMap.get(storageGroup)));
- }
- dataNodeLocations.put(index.getAndIncrement(), dataNodeLocation);
- });
- });
- });
- sendAsyncRequestToDataNodeWithRetry(req, handlerMap, dataNodeLocations);
+ CountDownLatch countDownLatch = new CountDownLatch(dataNodeLocations.size());
+ for (int retry = 0; retry < retryNum; retry++) {
+ AbstractRetryHandler handler = null;
+ for (Map.Entry<String, List<TRegionReplicaSet>> entry :
Review Comment:
The above has been modified.
But `entry.getKey()` needs to be used when creating the request here, so it has not been modified
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #6735: [IOTDB-3880] Optimize asyncclientpool retry logic
Posted by GitBox <gi...@apache.org>.
CRZbulabula commented on code in PR #6735:
URL: https://github.com/apache/iotdb/pull/6735#discussion_r927268169
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/AbstractRetryHandler.java:
##########
@@ -27,15 +27,15 @@
public abstract class AbstractRetryHandler {
- protected final int index;
-
protected CountDownLatch countDownLatch;
+
/**
- * Map<Index, TDataNodeLocation> The DataNode that successfully execute the request will be
- * removed from this map
+ * The DataNode that successfully execute the request will be removed from this list Because
+ * different requests will be sent to the same node when createRegions,so for CreateRegions ->
+ * Map<index, TDataNodeLocation> for others -> Map<dataNodeId, TDataNodeLocation>
Review Comment:
```suggestion
* Map<DataNodeId, TDataNodeLocation> The DataNode that successfully execute the request
* will be remoed from this map.
```
We can keep this annotation remains the same since the CreateRegionHandler is a particular case. And we can move this part of special comment into the CreateRegionHandler
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java:
##########
@@ -72,38 +71,63 @@ private AsyncDataNodeClientPool() {
* receive the requests
*
* @param req request
- * @param handlerMap Map<index, Handler>
- * @param dataNodeLocations ConcurrentHashMap<index, TDataNodeLocation> The specific DataNodes
+ * @param dataNodeLocationMap Map<DataNodeId, TDataNodeLocation>
+ * @param requestType DataNodeRequestType
+ * @param dataNodeResponseStatus response list.Used by CREATE_FUNCTION,DROP_FUNCTION and FLUSH
*/
public void sendAsyncRequestToDataNodeWithRetry(
Object req,
- Map<Integer, AbstractRetryHandler> handlerMap,
- Map<Integer, TDataNodeLocation> dataNodeLocations) {
- CountDownLatch countDownLatch = new CountDownLatch(dataNodeLocations.size());
- if (dataNodeLocations.isEmpty()) {
+ Map<Integer, TDataNodeLocation> dataNodeLocationMap,
+ DataNodeRequestType requestType,
+ List<TSStatus> dataNodeResponseStatus) {
+ if (dataNodeLocationMap.isEmpty()) {
return;
}
for (int retry = 0; retry < retryNum; retry++) {
- AbstractRetryHandler handler = null;
- for (Map.Entry<Integer, TDataNodeLocation> entry : dataNodeLocations.entrySet()) {
- handler = handlerMap.get(entry.getKey());
- // If it is not the first request, then prove that this operation is a retry.
- // The count of countDownLatch needs to be updated
- if (retry != 0) {
- handler.setCountDownLatch(countDownLatch);
+ CountDownLatch countDownLatch = new CountDownLatch(dataNodeLocationMap.size());
+ AbstractRetryHandler handler;
+ for (TDataNodeLocation targetDataNode : dataNodeLocationMap.values()) {
Review Comment:
It seems that this handler can be defined inside the following loop
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] RYH61 commented on a diff in pull request #6735: [IOTDB-3880] Optimize asyncclientpool retry logic
Posted by GitBox <gi...@apache.org>.
RYH61 commented on code in PR #6735:
URL: https://github.com/apache/iotdb/pull/6735#discussion_r926233170
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/CreateRegionHandler.java:
##########
@@ -38,23 +38,26 @@ public class CreateRegionHandler extends AbstractRetryHandler
private static final Logger LOGGER = LoggerFactory.getLogger(CreateRegionHandler.class);
+ private int index;
// Used for Logger
private final TConsensusGroupId consensusGroupId;
public CreateRegionHandler(
- int index,
CountDownLatch latch,
+ DataNodeRequestType requestType,
TConsensusGroupId consensusGroupId,
TDataNodeLocation targetDataNode,
- Map<Integer, TDataNodeLocation> dataNodeLocations) {
- super(latch, DataNodeRequestType.CREATE_REGIONS, targetDataNode, dataNodeLocations, index);
+ List<TDataNodeLocation> dataNodeLocations,
+ int index) {
+ super(latch, requestType, targetDataNode, dataNodeLocations);
+ this.index = index;
this.consensusGroupId = consensusGroupId;
}
@Override
public void onComplete(TSStatus tsStatus) {
if (tsStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- getDataNodeLocations().remove(index);
+ dataNodeLocations.remove(index);
Review Comment:
fixed~
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java:
##########
@@ -72,31 +72,63 @@ private AsyncDataNodeClientPool() {
* receive the requests
*
* @param req request
- * @param handlerMap Map<index, Handler>
- * @param dataNodeLocations ConcurrentHashMap<index, TDataNodeLocation> The specific DataNodes
+ * @param dataNodeLocations The specific DataNodes
+ * @param requestType DataNodeRequestType
+ * @param dataNodeResponseStatus request response list
Review Comment:
fixed~
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java:
##########
@@ -72,31 +72,63 @@ private AsyncDataNodeClientPool() {
* receive the requests
*
* @param req request
- * @param handlerMap Map<index, Handler>
- * @param dataNodeLocations ConcurrentHashMap<index, TDataNodeLocation> The specific DataNodes
+ * @param dataNodeLocations The specific DataNodes
+ * @param requestType DataNodeRequestType
+ * @param dataNodeResponseStatus request response list
*/
public void sendAsyncRequestToDataNodeWithRetry(
Object req,
- Map<Integer, AbstractRetryHandler> handlerMap,
- Map<Integer, TDataNodeLocation> dataNodeLocations) {
- CountDownLatch countDownLatch = new CountDownLatch(dataNodeLocations.size());
+ List<TDataNodeLocation> dataNodeLocations,
+ DataNodeRequestType requestType,
+ List<TSStatus> dataNodeResponseStatus) {
if (dataNodeLocations.isEmpty()) {
return;
}
+ CountDownLatch countDownLatch = new CountDownLatch(dataNodeLocations.size());
+ AbstractRetryHandler handler = null;
for (int retry = 0; retry < retryNum; retry++) {
- AbstractRetryHandler handler = null;
- for (Map.Entry<Integer, TDataNodeLocation> entry : dataNodeLocations.entrySet()) {
- handler = handlerMap.get(entry.getKey());
+ for (TDataNodeLocation targetDataNode : dataNodeLocations) {
+ switch (requestType) {
+ case SET_TTL:
+ handler =
+ new SetTTLHandler(countDownLatch, requestType, targetDataNode, dataNodeLocations);
+ break;
+ case CREATE_FUNCTION:
+ case DROP_FUNCTION:
Review Comment:
fixed~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org