add new interfaces to P2P Mode #330

lalalalalafeier · 2025-04-29T08:00:11Z

Summary

add new interfaces to support peer node and abort

Solution Description

CLAassistant · 2025-04-29T08:11:58Z

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.

sa-buc seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please add the email address used for this commit to your account.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

src/main/java/com/alipay/oceanbase/rpc/direct_load/execution/ObDirectLoadStatementExecutor.java

@@ -232,7 +257,11 @@ public void close() {
            }
        }
        // 退出任务
-        abortIfNeed();
+        if (isPeer) {
+            abortIfNeedWhenPeerClosed();


src/main/java/com/alipay/oceanbase/rpc/direct_load/ObDirectLoadStatement.java

+        executor.resumePeer(executionId, newTraceId);
+    }
+
+    public ObTableLoadClientStatus sendHeartBeat() throws ObDirectLoadException {


src/main/java/com/alipay/oceanbase/rpc/direct_load/ObDirectLoadStatement.java

@@ -304,6 +323,25 @@ public void resume(ObDirectLoadStatementExecutionId executionId) throws ObDirect
        executor.resume(executionId);
    }

+    public void resumePeer(ObDirectLoadStatementExecutionId executionId, ObDirectLoadTraceId newTraceId) throws ObDirectLoadException {


example/simple-table-demo/src/main/java/com/oceanbase/example/ObDirectLoadDemo.java

@@ -309,4 +315,303 @@ public static void run() {

    };

+    private static class MultiPeerNodeWriteTest {


suz-yang · 2025-04-29T08:56:16Z

example/simple-table-demo/src/main/java/com/oceanbase/example/ObDirectLoadDemo.java

+                connection = buildConnection(1);
+                statement = buildStatement(connection);
+
+                SendHeartBeat sendHeartBeat = new SendHeartBeat(connection, statement);


不应该在SendHeartBeat里面再build一个新的stmt去发心跳么？

src/main/java/com/alipay/oceanbase/rpc/direct_load/ObDirectLoadStatement.java

@@ -251,6 +252,24 @@ public void begin() throws ObDirectLoadException {
        }
    }

+    public ObDirectLoadStatementFuture beginPeerAsync() {


src/main/java/com/alipay/oceanbase/rpc/direct_load/ObDirectLoadStatement.java

@@ -279,6 +282,10 @@ public void write(ObDirectLoadBucket bucket) throws ObDirectLoadException {
        executor.write(bucket);
    }

+    public void startHeartBeat() throws ObDirectLoadException {
+        executor.startHeartBeatInP2PMode();


src/main/java/com/alipay/oceanbase/rpc/direct_load/ObDirectLoadStatement.java

@@ -377,6 +389,11 @@ public ObDirectLoadStatement build() throws ObDirectLoadException {
            return connection.buildStatement(this);
        }

+        public Builder setP2PMode() {


src/main/java/com/alipay/oceanbase/rpc/direct_load/execution/ObDirectLoadStatementExecutor.java

@@ -355,6 +419,10 @@ public void write(ObDirectLoadBucket bucket) throws ObDirectLoadException {
        }
    }

+    public void execAbort() throws ObDirectLoadException {
+        abort();


src/main/java/com/alipay/oceanbase/rpc/direct_load/ObDirectLoadStatement.java

@@ -382,6 +388,11 @@ public ObDirectLoadTraceId getTraceId() {
            return traceId;
        }

+        public Builder setNodeRole(NodeRole nodeRole) {


src/main/java/com/alipay/oceanbase/rpc/direct_load/execution/ObDirectLoadStatementExecutor.java

@@ -317,7 +336,7 @@ private synchronized void abortIfNeed() {
        }
    }

-    private ObDirectLoadStatementFuture abort() {
+    public ObDirectLoadStatementFuture abort() {


example/simple-table-demo/src/main/java/com/oceanbase/example/ObDirectLoadDemo.java

+                    Thread thread = new Thread(NodeWriter);
+                    threads[i] = thread;
+                }
+                P2PNodeWriter commitNodeWriter = new P2PNodeWriter(executionIdBytes, writeThreadNum, threads);


src/main/java/com/alipay/oceanbase/rpc/direct_load/execution/ObDirectLoadStatementExecutor.java

@@ -58,13 +58,19 @@ public class ObDirectLoadStatementExecutor {
    private long                               taskId        = 0;
    private ObAddr                             svrAddr       = null;
    private ObDirectLoadException              cause         = null;                   // 失败原因
+    private NodeRole                           nodeRole      = NodeRole.PRIMARY;


src/main/java/com/alipay/oceanbase/rpc/direct_load/execution/ObDirectLoadStatementExecutor.java

-            compareAndSetState(NONE, LOADING_ONLY, "resume");
+            if (NodeRole.P2P == nodeRole) {
+                compareAndSetState(NONE, LOADING, "resume in P2P mode");
+                startHeartBeat();


suz-yang · 2025-06-13T03:56:29Z

src/main/java/com/alipay/oceanbase/rpc/direct_load/ObDirectLoadStatement.java

@@ -327,6 +331,7 @@ public static final class Builder {
        private ObDirectLoadStatementExecutionId executionId       = null;

        private static final long                MAX_QUERY_TIMEOUT = Integer.MAX_VALUE;
+        private boolean                          isP2PMode         = false;


为什么加到这个位置？

suz-yang · 2025-06-13T03:58:52Z

src/main/java/com/alipay/oceanbase/rpc/direct_load/execution/ObDirectLoadStatementExecutor.java

@@ -123,6 +141,16 @@ public synchronized ObDirectLoadStatementFuture begin() {
    public synchronized ObDirectLoadStatementFuture commit() {
        logger.info("statement call commit");
        ObDirectLoadStatementAsyncPromiseTask task = null;
+        try {


为啥要套一层这个？

suz-yang reviewed Apr 29, 2025

View reviewed changes

src/main/java/com/alipay/oceanbase/rpc/direct_load/execution/ObDirectLoadStatementExecutor.java Outdated

@@ -232,7 +257,11 @@ public void close() {

}

}

// 退出任务

abortIfNeed();

if (isPeer) {

abortIfNeedWhenPeerClosed();

This comment was marked as resolved.

Sign in to view

suz-yang reviewed Apr 29, 2025

View reviewed changes

src/main/java/com/alipay/oceanbase/rpc/direct_load/ObDirectLoadStatement.java Outdated

executor.resumePeer(executionId, newTraceId);

}

public ObTableLoadClientStatus sendHeartBeat() throws ObDirectLoadException {

This comment was marked as resolved.

Sign in to view

suz-yang reviewed Apr 29, 2025

View reviewed changes

src/main/java/com/alipay/oceanbase/rpc/direct_load/ObDirectLoadStatement.java Outdated

@@ -251,6 +252,24 @@ public void begin() throws ObDirectLoadException {

}

}

public ObDirectLoadStatementFuture beginPeerAsync() {

This comment was marked as resolved.

Sign in to view

lalalalalafeier force-pushed the master branch from 3a380e2 to ab5a11b Compare April 30, 2025 07:42

lalalalalafeier changed the title ~~add new interfaces to support peer node and abort~~ add new interfaces to P2P Mode Apr 30, 2025

suz-yang reviewed Apr 30, 2025

View reviewed changes

lalalalalafeier force-pushed the master branch from ab5a11b to 183e8bb Compare May 6, 2025 07:38

suz-yang reviewed May 6, 2025

View reviewed changes

src/main/java/com/alipay/oceanbase/rpc/direct_load/ObDirectLoadStatement.java Outdated

@@ -382,6 +388,11 @@ public ObDirectLoadTraceId getTraceId() {

return traceId;

}

public Builder setNodeRole(NodeRole nodeRole) {

This comment was marked as resolved.

Sign in to view

suz-yang reviewed May 6, 2025

View reviewed changes

suz-yang reviewed May 14, 2025

View reviewed changes

lalalalalafeier force-pushed the master branch 3 times, most recently from e1d7a8e to 24c79d5 Compare May 27, 2025 09:06

suz-yang reviewed Jun 13, 2025

View reviewed changes

add P2P mode and abort interface

63e205b

lalalalalafeier force-pushed the master branch from 24c79d5 to 63e205b Compare June 20, 2025 12:06

		@@ -309,4 +315,303 @@ public static void run() {

		};

		private static class MultiPeerNodeWriteTest {

add new interfaces to P2P Mode #330

Are you sure you want to change the base?

add new interfaces to P2P Mode #330

Uh oh!

Conversation

lalalalalafeier commented Apr 29, 2025

Summary

Solution Description

Uh oh!

CLAassistant commented Apr 29, 2025

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

suz-yang Apr 29, 2025

Choose a reason for hiding this comment

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

suz-yang Jun 13, 2025

Choose a reason for hiding this comment

Uh oh!

suz-yang Jun 13, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!