You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ta...@apache.org on 2021/07/09 03:33:19 UTC
[iotdb] branch fix_cluster created (now f8429d9)
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a change to branch fix_cluster
in repository https://gitbox.apache.org/repos/asf/iotdb.git.
at f8429d9 fix
This branch includes the following new commits:
new f8429d9 fix
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/01: fix
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch fix_cluster
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit f8429d9f9664bb24323a86e8d1c6a51046cf8f19
Author: LebronAl <TX...@gmail.com>
AuthorDate: Fri Jul 9 11:32:36 2021 +0800
fix
---
.../apache/iotdb/cluster/coordinator/Coordinator.java | 6 +++---
.../org/apache/iotdb/cluster/metadata/CMManager.java | 5 ++++-
.../cluster/partition/slot/SlotPartitionTable.java | 3 ++-
.../iotdb/cluster/server/MetaClusterServer.java | 1 +
.../java/org/apache/iotdb/db/metadata/MManager.java | 19 ++++++++++++++-----
.../main/java/org/apache/iotdb/db/service/IoTDB.java | 9 +++++++++
.../main/java/org/apache/iotdb/session/Session.java | 2 +-
7 files changed, 34 insertions(+), 11 deletions(-)
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/coordinator/Coordinator.java b/cluster/src/main/java/org/apache/iotdb/cluster/coordinator/Coordinator.java
index aa168d5..11f99e8 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/coordinator/Coordinator.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/coordinator/Coordinator.java
@@ -385,10 +385,10 @@ public class Coordinator {
private TSStatus createTimeseriesForFailedInsertion(
Map<PhysicalPlan, PartitionGroup> planGroupMap, InsertPlan plan) {
- // try to create timeseries
- if (plan.getFailedMeasurements() != null) {
- plan.getPlanFromFailed();
+ for (PhysicalPlan subPlan : planGroupMap.keySet()) {
+ ((InsertPlan) subPlan).recoverFromFailure();
}
+ // try to create timeseries
boolean hasCreate;
try {
hasCreate = ((CMManager) IoTDB.metaManager).createTimeseries(plan);
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java b/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java
index 12c1116..8f034c6 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java
@@ -732,7 +732,10 @@ public class CMManager extends MManager {
}
if (result.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()
&& result.getCode() != TSStatusCode.PATH_ALREADY_EXIST_ERROR.getStatusCode()
- && result.getCode() != TSStatusCode.NEED_REDIRECTION.getStatusCode()) {
+ && result.getCode() != TSStatusCode.NEED_REDIRECTION.getStatusCode()
+ && result.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()
+ && result.getSubStatus().stream()
+ .allMatch(s -> s.getCode() == TSStatusCode.PATH_ALREADY_EXIST_ERROR.getStatusCode())) {
logger.error(
"{} failed to execute create timeseries {}: {}",
metaGroupMember.getThisNode(),
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/partition/slot/SlotPartitionTable.java b/cluster/src/main/java/org/apache/iotdb/cluster/partition/slot/SlotPartitionTable.java
index 469e84d..571a7b9 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/partition/slot/SlotPartitionTable.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/partition/slot/SlotPartitionTable.java
@@ -25,6 +25,7 @@ import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
@@ -103,7 +104,7 @@ public class SlotPartitionTable implements PartitionTable {
private void init(Collection<Node> nodes) {
logger.info("Initializing a new partition table");
nodeRing.addAll(nodes);
- nodeRing.sort(Comparator.comparingInt(Node::getNodeIdentifier));
+ Collections.sort(nodeRing);
localGroups = getPartitionGroups(thisNode);
assignPartitions();
}
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/server/MetaClusterServer.java b/cluster/src/main/java/org/apache/iotdb/cluster/server/MetaClusterServer.java
index 12e286f..abb8020 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/server/MetaClusterServer.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/server/MetaClusterServer.java
@@ -108,6 +108,7 @@ public class MetaClusterServer extends RaftServer
metaHeartbeatServer.start();
ioTDB = new IoTDB();
IoTDB.setMetaManager(CMManager.getInstance());
+ IoTDB.setClusterMode();
((CMManager) IoTDB.metaManager).setMetaGroupMember(member);
((CMManager) IoTDB.metaManager).setCoordinator(coordinator);
ioTDB.active();
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
index ba434bc..072ecdd 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
@@ -64,6 +64,7 @@ import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.dataset.ShowDevicesResult;
import org.apache.iotdb.db.query.dataset.ShowTimeSeriesResult;
import org.apache.iotdb.db.rescon.MemTableManager;
+import org.apache.iotdb.db.service.IoTDB;
import org.apache.iotdb.db.utils.RandomDeleteCache;
import org.apache.iotdb.db.utils.SchemaUtils;
import org.apache.iotdb.db.utils.TestOnly;
@@ -1937,11 +1938,19 @@ public class MManager {
measurementList[i] = measurementMNode.getName();
} catch (MetadataException e) {
- logger.warn(
- "meet error when check {}.{}, message: {}",
- deviceId,
- measurementList[i],
- e.getMessage());
+ if (IoTDB.isClusterMode()) {
+ logger.debug(
+ "meet error when check {}.{}, message: {}",
+ deviceId,
+ measurementList[i],
+ e.getMessage());
+ } else {
+ logger.warn(
+ "meet error when check {}.{}, message: {}",
+ deviceId,
+ measurementList[i],
+ e.getMessage());
+ }
if (config.isEnablePartialInsert()) {
// mark failed measurement
plan.markFailedMeasurementInsertion(i, e);
diff --git a/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java b/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
index 6a516db..a226379 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
@@ -53,6 +53,7 @@ public class IoTDB implements IoTDBMBean {
String.format("%s:%s=%s", IoTDBConstant.IOTDB_PACKAGE, IoTDBConstant.JMX_TYPE, "IoTDB");
private RegisterManager registerManager = new RegisterManager();
public static MManager metaManager = MManager.getInstance();
+ public static boolean isClusterMode = false;
public static IoTDB getInstance() {
return IoTDBHolder.INSTANCE;
@@ -72,6 +73,14 @@ public class IoTDB implements IoTDBMBean {
IoTDB.metaManager = metaManager;
}
+ public static void setClusterMode() {
+ IoTDB.isClusterMode = true;
+ }
+
+ public static boolean isClusterMode() {
+ return IoTDB.isClusterMode;
+ }
+
public void active() {
StartupChecks checks = new StartupChecks().withDefaultTest();
try {
diff --git a/session/src/main/java/org/apache/iotdb/session/Session.java b/session/src/main/java/org/apache/iotdb/session/Session.java
index 9fbcb1e..183137a 100644
--- a/session/src/main/java/org/apache/iotdb/session/Session.java
+++ b/session/src/main/java/org/apache/iotdb/session/Session.java
@@ -564,7 +564,7 @@ public class Session {
private SessionDataSet executeStatementMayRedirect(String sql, long timeoutInMs)
throws StatementExecutionException, IoTDBConnectionException {
try {
- logger.info("{} execute sql {}", defaultSessionConnection.getEndPoint(), sql);
+ logger.debug("{} execute sql {}", defaultSessionConnection.getEndPoint(), sql);
return defaultSessionConnection.executeQueryStatement(sql, timeoutInMs);
} catch (RedirectException e) {
handleQueryRedirection(e.getEndPoint());