Skip to content
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

Pipe: using clusterId to judge whether the target cluster is source cluster #11994

Merged
merged 24 commits into from
Feb 2, 2024
Merged

Pipe: using clusterId to judge whether the target cluster is source cluster #11994

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
24 commits
Select commit Hold shift + click to select a range
694cd6e
update a new handshake request
Jan 28, 2024
7ac2296
format code
Jan 28, 2024
248ae19
1. rename the PipeTransferHandshakeReq to PipeTransferHandshakeV1Req.
Jan 29, 2024
ff01ee9
handle PipeTransferHandshakeV2Req in sync connector
Jan 29, 2024
ab1419d
1. send PipeTransferHandshakeV2Req default in async connector
Jan 29, 2024
b71d031
add PipeConstant class
Jan 29, 2024
c4577c9
format code
Jan 29, 2024
2a91d77
fix problems of comments
Jan 30, 2024
dbc53d2
Merge branch 'master' of https://github.com/apache/iotdb into TIMECHO…
Jan 30, 2024
0c1d93c
add apache rat in PipeTransferHandshakeV2Req
Jan 30, 2024
288419d
fix problems of comments.
Jan 31, 2024
d975937
rename PipeConstant to PipeConnectorConstant
Jan 31, 2024
c3ef4aa
format code
Jan 31, 2024
4c04878
Merge branch 'master' of https://github.com/apache/iotdb into pr/11994
SteveYurongSu Feb 1, 2024
1c3d981
fix
SteveYurongSu Feb 1, 2024
cb63937
refactor
SteveYurongSu Feb 2, 2024
47edecf
RENAME
SteveYurongSu Feb 2, 2024
68c380b
fix receiver
SteveYurongSu Feb 2, 2024
4cb7011
Update IoTDBThriftReceiverV1.java
SteveYurongSu Feb 2, 2024
f9a1d71
Update IoTDBThriftSyncClientManager.java
SteveYurongSu Feb 2, 2024
f5f79f8
Update IoTDBThriftAsyncClientManager.java
SteveYurongSu Feb 2, 2024
d6db5f1
Update IoTDBAirGapConnector.java
SteveYurongSu Feb 2, 2024
075f8bc
refactor
SteveYurongSu Feb 2, 2024
7efd38f
Update PipeTransferHandshakeV2Req.java
SteveYurongSu Feb 2, 2024
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
Prev Previous commit
Next Next commit
refactor
  • Loading branch information
@SteveYurongSu
SteveYurongSu committed Feb 2, 2024
commit cb63937b2c880289ea09a77e4a287e8655e13138
38 changes: 20 additions & 18 deletions ...-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeRuntimeAgent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,35 +44,21 @@

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

public class PipeRuntimeAgent implements IService {

private static final Logger LOGGER = LoggerFactory.getLogger(PipeRuntimeAgent.class);
private static final int DATA_NODE_ID = IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
private static volatile String clusterId = null;

private final AtomicBoolean isShutdown = new AtomicBoolean(false);

private final PipePeriodicalJobExecutor pipePeriodicalJobExecutor =
new PipePeriodicalJobExecutor();
private final AtomicReference<String> clusterId = new AtomicReference<>(null);

private final SimpleConsensusProgressIndexAssigner simpleConsensusProgressIndexAssigner =
new SimpleConsensusProgressIndexAssigner();

//////////////////////////// Getter ////////////////////////////

public static synchronized String getClusterId() {
if (clusterId == null) {
try (ConfigNodeClient configNodeClient =
ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
clusterId = configNodeClient.getClusterId().clusterId;
} catch (Exception e) {
LOGGER.warn("Unable to get clusterId, because: {}", e.getMessage(), e);
}
}

return clusterId;
}
private final PipePeriodicalJobExecutor pipePeriodicalJobExecutor =
new PipePeriodicalJobExecutor();

//////////////////////////// System Service Interface ////////////////////////////

Expand Down Expand Up @@ -122,6 +108,22 @@ public ServiceType getID() {
return ServiceType.PIPE_RUNTIME_AGENT;
}

public String getClusterIdIfPossible() {
if (clusterId.get() == null) {
synchronized (clusterId) {
if (clusterId.get() == null) {
try (final ConfigNodeClient configNodeClient =
ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
clusterId.set(configNodeClient.getClusterId().getClusterId());
} catch (Exception e) {
LOGGER.warn("Unable to get clusterId, because: {}", e.getMessage(), e);
}
}
}
}
return clusterId.get();
}

////////////////////// SimpleConsensus ProgressIndex Assigner //////////////////////

public void assignSimpleProgressIndexIfNeeded(InsertNode insertNode) {
Expand Down
6 changes: 4 additions & 2 deletions ...rc/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBAirGapConnector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@
import org.apache.iotdb.commons.utils.NodeUrlUtils;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.agent.runtime.PipeRuntimeAgent;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
import org.apache.iotdb.db.pipe.connector.payload.airgap.AirGapELanguageConstant;
import org.apache.iotdb.db.pipe.connector.payload.airgap.AirGapOneByteResponse;
import org.apache.iotdb.db.pipe.connector.payload.evolvable.common.PipeConnectorConstant;
Expand Down Expand Up @@ -200,7 +200,9 @@ public void handshake() throws Exception {

// Try to handshake by PipeTransferHandshakeV2Req.
HashMap<String, String> params = new HashMap<>();
params.put(PipeConnectorConstant.HANDSHAKE_KEY_CLUSTER_ID, PipeRuntimeAgent.getClusterId());
params.put(
PipeConnectorConstant.HANDSHAKE_KEY_CLUSTER_ID,
PipeAgent.runtime().getClusterIdIfPossible());
params.put(
PipeConnectorConstant.HANDSHAKE_KEY_TIME_PRECISION,
CommonDescriptor.getInstance().getConfig().getTimestampPrecision());
Expand Down
6 changes: 4 additions & 2 deletions ...g/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncClientManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.db.pipe.agent.runtime.PipeRuntimeAgent;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
import org.apache.iotdb.db.pipe.connector.payload.evolvable.common.PipeConnectorConstant;
import org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferHandshakeV1Req;
import org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferHandshakeV2Req;
Expand Down Expand Up @@ -175,7 +175,9 @@ public void onError(Exception e) {

// Try to handshake by PipeTransferHandshakeV2Req.
HashMap<String, String> params = new HashMap<>();
params.put(PipeConnectorConstant.HANDSHAKE_KEY_CLUSTER_ID, PipeRuntimeAgent.getClusterId());
params.put(
PipeConnectorConstant.HANDSHAKE_KEY_CLUSTER_ID,
PipeAgent.runtime().getClusterIdIfPossible());
params.put(
PipeConnectorConstant.HANDSHAKE_KEY_TIME_PRECISION,
CommonDescriptor.getInstance().getConfig().getTimestampPrecision());
Expand Down
6 changes: 4 additions & 2 deletions ...org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncClientManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.connector.client.IoTDBThriftSyncConnectorClient;
import org.apache.iotdb.db.pipe.agent.runtime.PipeRuntimeAgent;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
import org.apache.iotdb.db.pipe.connector.payload.evolvable.common.PipeConnectorConstant;
import org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferHandshakeV1Req;
import org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferHandshakeV2Req;
Expand Down Expand Up @@ -152,7 +152,9 @@ public void sendHandshakeReq(
params.put(
PipeConnectorConstant.HANDSHAKE_KEY_TIME_PRECISION,
CommonDescriptor.getInstance().getConfig().getTimestampPrecision());
params.put(PipeConnectorConstant.HANDSHAKE_KEY_CLUSTER_ID, PipeRuntimeAgent.getClusterId());
params.put(
PipeConnectorConstant.HANDSHAKE_KEY_CLUSTER_ID,
PipeAgent.runtime().getClusterIdIfPossible());

TPipeTransferResp resp =
clientAndStatus
Expand Down
4 changes: 2 additions & 2 deletions ...atanode/src/main/java/org/apache/iotdb/db/pipe/receiver/thrift/IoTDBThriftReceiverV1.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.DiskSpaceInsufficientException;
import org.apache.iotdb.db.pipe.agent.runtime.PipeRuntimeAgent;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
import org.apache.iotdb.db.pipe.connector.payload.airgap.AirGapPseudoTPipeTransferRequest;
import org.apache.iotdb.db.pipe.connector.payload.evolvable.common.PipeConnectorConstant;
import org.apache.iotdb.db.pipe.connector.payload.evolvable.reponse.PipeTransferFilePieceResp;
Expand Down Expand Up @@ -254,7 +254,7 @@ private TPipeTransferResp handleTransferHandshakeV1(PipeTransferHandshakeV1Req r
private TPipeTransferResp handleTransferHandshakeV2(PipeTransferHandshakeV2Req req)
throws IOException {
// Reject to handshake if the receiver can not take clusterId from configNode.
String clusterId = PipeRuntimeAgent.getClusterId();
String clusterId = PipeAgent.runtime().getClusterIdIfPossible();
if (clusterId == null) {
final TSStatus status =
RpcUtils.getStatus(
Expand Down
Loading