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/08/01 10:04:06 UTC
[GitHub] [iotdb] ljn55966005 opened a new pull request, #6855: Rel/0.13 secondary
ljn55966005 opened a new pull request, #6855:
URL: https://github.com/apache/iotdb/pull/6855
## change the secondary version to meet some needs
### add the queue number
### add the heartbeat check
### change the dml process
### the queue turns to be not block
--
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 #6855: [To rel/0.13] Improve the process of secondary
Posted by GitBox <gi...@apache.org>.
CRZbulabula commented on code in PR #6855:
URL: https://github.com/apache/iotdb/pull/6855#discussion_r936416085
##########
server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java:
##########
@@ -87,33 +86,17 @@
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Paths;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Comparator;
-import java.util.ConcurrentModificationException;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
Review Comment:
We should always avoid to `import *`
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncPlanTypeUtils.java:
##########
@@ -21,21 +21,7 @@
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
-import org.apache.iotdb.db.qp.physical.sys.ActivateTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.AlterTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.AppendTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateMultiTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.DeactivateTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.DeleteStorageGroupPlan;
-import org.apache.iotdb.db.qp.physical.sys.DeleteTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.DropTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.PruneTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
-import org.apache.iotdb.db.qp.physical.sys.SetTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.UnsetTemplatePlan;
+import org.apache.iotdb.db.qp.physical.sys.*;
Review Comment:
We should always avoid to `import *`
##########
server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java:
##########
@@ -87,33 +86,17 @@
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Paths;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Comparator;
-import java.util.ConcurrentModificationException;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
import java.util.Map.Entry;
-import java.util.Set;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
Review Comment:
We should always avoid to `import *`
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncProducer.java:
##########
@@ -33,22 +35,52 @@
public class OperationSyncProducer {
private static final Logger LOGGER = LoggerFactory.getLogger(OperationSyncProducer.class);
-
- private final BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>
- operationSyncQueue;
+ private static final Integer RETRY = 3;
+ private final ArrayList<
+ BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>>
+ operationSyncQueues;
+ private final OperationSyncLogService dmlLogService;
public OperationSyncProducer(
- BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>
- operationSyncQueue) {
- this.operationSyncQueue = operationSyncQueue;
+ ArrayList<BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>>
+ operationSyncQueue,
+ OperationSyncLogService operationSyncDMLLogService) {
+ this.operationSyncQueues = operationSyncQueue;
+ this.dmlLogService = operationSyncDMLLogService;
}
- public void put(Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType> planPair) {
+ public void put(
+ Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType> planPair,
+ String deviceName) {
+
+ ByteBuffer headBuffer;
+ headBuffer = planPair.left;
+ headBuffer.position(0);
+ int index;
+ if (deviceName == null) {
+ index = operationSyncQueues.size() - 1;
+ } else {
+ try {
+ index = Math.abs(deviceName.hashCode()) % (operationSyncQueues.size() - 1);
+ } catch (Exception e) {
+ index = 1;
+ }
+ }
+ for (int i = 0; i < RETRY; i++) {
+ // retry 3 times
+ if (operationSyncQueues.get(index).offer(planPair)) {
+ return;
+ }
+ }
Review Comment:
It's better to sleep 1s or less between each loop of retry.
##########
server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java:
##########
@@ -58,21 +58,20 @@
import org.apache.iotdb.db.metadata.mnode.IStorageGroupMNode;
import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
-import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
-import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
-import org.apache.iotdb.db.qp.physical.crud.InsertRowsOfOneDevicePlan;
-import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
+import org.apache.iotdb.db.qp.physical.crud.*;
Review Comment:
We should always avoid to `import *`
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncProducer.java:
##########
@@ -33,22 +35,52 @@
public class OperationSyncProducer {
private static final Logger LOGGER = LoggerFactory.getLogger(OperationSyncProducer.class);
-
- private final BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>
- operationSyncQueue;
+ private static final Integer RETRY = 3;
+ private final ArrayList<
+ BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>>
+ operationSyncQueues;
+ private final OperationSyncLogService dmlLogService;
public OperationSyncProducer(
- BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>
- operationSyncQueue) {
- this.operationSyncQueue = operationSyncQueue;
+ ArrayList<BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>>
+ operationSyncQueue,
+ OperationSyncLogService operationSyncDMLLogService) {
+ this.operationSyncQueues = operationSyncQueue;
+ this.dmlLogService = operationSyncDMLLogService;
}
- public void put(Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType> planPair) {
+ public void put(
+ Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType> planPair,
+ String deviceName) {
+
+ ByteBuffer headBuffer;
+ headBuffer = planPair.left;
+ headBuffer.position(0);
+ int index;
+ if (deviceName == null) {
+ index = operationSyncQueues.size() - 1;
+ } else {
+ try {
+ index = Math.abs(deviceName.hashCode()) % (operationSyncQueues.size() - 1);
+ } catch (Exception e) {
+ index = 1;
+ }
Review Comment:
I wonder in which case will this block of code throw an Exception? And the same as before, would it be better if we use random selection when some Exceptions are thrown?
##########
server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java:
##########
@@ -122,6 +105,7 @@ public class StorageEngine implements IService {
private static OperationSyncProducer operationSyncProducer;
private static OperationSyncDDLProtector operationSyncDDLProtector;
private static OperationSyncLogService operationSyncDDLLogService;
+ private static boolean isSecondaryLife;
Review Comment:
This field should better to be set to AtomicBoolean or volatile for thread-safety
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncProducer.java:
##########
@@ -33,22 +35,52 @@
public class OperationSyncProducer {
private static final Logger LOGGER = LoggerFactory.getLogger(OperationSyncProducer.class);
-
- private final BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>
- operationSyncQueue;
+ private static final Integer RETRY = 3;
+ private final ArrayList<
+ BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>>
+ operationSyncQueues;
+ private final OperationSyncLogService dmlLogService;
public OperationSyncProducer(
- BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>
- operationSyncQueue) {
- this.operationSyncQueue = operationSyncQueue;
+ ArrayList<BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>>
+ operationSyncQueue,
+ OperationSyncLogService operationSyncDMLLogService) {
+ this.operationSyncQueues = operationSyncQueue;
+ this.dmlLogService = operationSyncDMLLogService;
}
- public void put(Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType> planPair) {
+ public void put(
+ Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType> planPair,
+ String deviceName) {
+
+ ByteBuffer headBuffer;
+ headBuffer = planPair.left;
+ headBuffer.position(0);
+ int index;
+ if (deviceName == null) {
+ index = operationSyncQueues.size() - 1;
Review Comment:
Would it be better if we use random selection here?
--
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] ljn55966005 commented on a diff in pull request #6855: [To rel/0.13] Improve the process of secondary
Posted by GitBox <gi...@apache.org>.
ljn55966005 commented on code in PR #6855:
URL: https://github.com/apache/iotdb/pull/6855#discussion_r936515917
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncProducer.java:
##########
@@ -33,22 +35,52 @@
public class OperationSyncProducer {
private static final Logger LOGGER = LoggerFactory.getLogger(OperationSyncProducer.class);
-
- private final BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>
- operationSyncQueue;
+ private static final Integer RETRY = 3;
+ private final ArrayList<
+ BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>>
+ operationSyncQueues;
+ private final OperationSyncLogService dmlLogService;
public OperationSyncProducer(
- BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>
- operationSyncQueue) {
- this.operationSyncQueue = operationSyncQueue;
+ ArrayList<BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>>
+ operationSyncQueue,
+ OperationSyncLogService operationSyncDMLLogService) {
+ this.operationSyncQueues = operationSyncQueue;
+ this.dmlLogService = operationSyncDMLLogService;
}
- public void put(Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType> planPair) {
+ public void put(
+ Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType> planPair,
+ String deviceName) {
+
+ ByteBuffer headBuffer;
+ headBuffer = planPair.left;
+ headBuffer.position(0);
+ int index;
+ if (deviceName == null) {
+ index = operationSyncQueues.size() - 1;
Review Comment:
we must prove the order of the same device
--
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] ljn55966005 commented on a diff in pull request #6855: [To rel/0.13] Improve the process of secondary
Posted by GitBox <gi...@apache.org>.
ljn55966005 commented on code in PR #6855:
URL: https://github.com/apache/iotdb/pull/6855#discussion_r936568745
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncProducer.java:
##########
@@ -33,22 +35,52 @@
public class OperationSyncProducer {
private static final Logger LOGGER = LoggerFactory.getLogger(OperationSyncProducer.class);
-
- private final BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>
- operationSyncQueue;
+ private static final Integer RETRY = 3;
+ private final ArrayList<
+ BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>>
+ operationSyncQueues;
+ private final OperationSyncLogService dmlLogService;
public OperationSyncProducer(
- BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>
- operationSyncQueue) {
- this.operationSyncQueue = operationSyncQueue;
+ ArrayList<BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>>
+ operationSyncQueue,
+ OperationSyncLogService operationSyncDMLLogService) {
+ this.operationSyncQueues = operationSyncQueue;
+ this.dmlLogService = operationSyncDMLLogService;
}
- public void put(Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType> planPair) {
+ public void put(
+ Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType> planPair,
+ String deviceName) {
+
+ ByteBuffer headBuffer;
+ headBuffer = planPair.left;
+ headBuffer.position(0);
+ int index;
+ if (deviceName == null) {
+ index = operationSyncQueues.size() - 1;
+ } else {
+ try {
+ index = Math.abs(deviceName.hashCode()) % (operationSyncQueues.size() - 1);
+ } catch (Exception e) {
+ index = 1;
+ }
Review Comment:
when the queue size is 1
--
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] qiaojialin merged pull request #6855: [To rel/0.13] Improve the process of secondary
Posted by GitBox <gi...@apache.org>.
qiaojialin merged PR #6855:
URL: https://github.com/apache/iotdb/pull/6855
--
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 #6855: Rel/0.13 secondary
Posted by GitBox <gi...@apache.org>.
Beyyes commented on code in PR #6855:
URL: https://github.com/apache/iotdb/pull/6855#discussion_r935183662
##########
server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java:
##########
@@ -122,6 +106,7 @@ public class StorageEngine implements IService {
private static OperationSyncProducer operationSyncProducer;
private static OperationSyncDDLProtector operationSyncDDLProtector;
private static OperationSyncLogService operationSyncDDLLogService;
+ private static boolean isSecondaryLife;
Review Comment:
`isSecondaryLife` change to `isSecondaryAlive`
##########
server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java:
##########
@@ -114,6 +97,7 @@ public class StorageEngine implements IService {
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
private static final long TTL_CHECK_INTERVAL = 60 * 1000L;
+ private static final long HEARTBEAT_CHECK_INTERVAL = 30L;
Review Comment:
If it's too long to check the alive info of Secondary in every 30 seconds
##########
server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java:
##########
@@ -393,6 +404,14 @@ public void start() {
startTimedService();
}
+ private void checkSecondaryIsLife() {
Review Comment:
```suggestion
private void checkSecondaryAlive() {
```
##########
server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java:
##########
@@ -122,6 +106,7 @@ public class StorageEngine implements IService {
private static OperationSyncProducer operationSyncProducer;
private static OperationSyncDDLProtector operationSyncDDLProtector;
private static OperationSyncLogService operationSyncDDLLogService;
+ private static boolean isSecondaryLife;
Review Comment:
```suggestion
private static boolean isSecondaryAlive;
```
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncPlanTypeUtils.java:
##########
@@ -21,21 +21,7 @@
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
-import org.apache.iotdb.db.qp.physical.sys.ActivateTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.AlterTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.AppendTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateMultiTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.DeactivateTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.DeleteStorageGroupPlan;
-import org.apache.iotdb.db.qp.physical.sys.DeleteTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.DropTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.PruneTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
-import org.apache.iotdb.db.qp.physical.sys.SetTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.UnsetTemplatePlan;
+import org.apache.iotdb.db.qp.physical.sys.*;
Review Comment:
using '*' is not recommended
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncConsumer.java:
##########
@@ -39,10 +40,10 @@ public class OperationSyncConsumer implements Runnable {
public OperationSyncConsumer(
BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>
- OperationSyncQueue,
+ operationSyncQueue,
SessionPool operationSyncSessionPool,
OperationSyncLogService dmlLogService) {
- this.OperationSyncQueue = OperationSyncQueue;
+ this.OperationSyncQueue = operationSyncQueue;
Review Comment:
this.operationSyncQueue
The first letter should be lower case.
##########
server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java:
##########
@@ -393,6 +404,14 @@ public void start() {
startTimedService();
}
+ private void checkSecondaryIsLife() {
Review Comment:
`life` is a noun
##########
server/src/main/java/org/apache/iotdb/db/doublelive/OperationSyncProducer.java:
##########
@@ -33,22 +35,47 @@
public class OperationSyncProducer {
private static final Logger LOGGER = LoggerFactory.getLogger(OperationSyncProducer.class);
-
- private final BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>
- operationSyncQueue;
+ private static final Integer RETRY = 3;
+ private final ArrayList<
+ BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>>
+ operationSyncQueues;
+ private final OperationSyncLogService dmlLogService;
public OperationSyncProducer(
- BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>
- operationSyncQueue) {
- this.operationSyncQueue = operationSyncQueue;
+ ArrayList<BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>>>
+ operationSyncQueue,
+ OperationSyncLogService operationSyncDMLLogService) {
+ this.operationSyncQueues = operationSyncQueue;
+ this.dmlLogService = operationSyncDMLLogService;
}
- public void put(Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType> planPair) {
+ public void put(
+ Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType> planPair,
+ String deviceName) {
+ ByteBuffer headBuffer;
+ headBuffer = planPair.left;
+ headBuffer.position(0);
+ int index;
+ if (deviceName == null) {
+ index = operationSyncQueues.size();
+ } else {
+ index = deviceName.hashCode() % (operationSyncQueues.size() - 1);
Review Comment:
The result of deviceName.hashCode() may be negative
--
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