fix: prevent sender control future race

[lokidundun]

Purpose of the PR

• close [Task] Investigate slow integrate-test stalls in message sender barrier #350

Main Changes

• Fixes a race condition in QueuedMessageSender where consecutive control messages could fail with The origin future must be null.

• The sender now clears the in-flight control future before completing it, allowing the next control message to be sent safely. Transport failures also complete the affected future exceptionally.

Tests

• Added regression tests for consecutive control messages and transport failures.
• Added fail-fast timeout handling to sender integration tests.
• Verified with QueuedMessageSenderTest and IntegrateTestSuite.

mvn test -P integrate-test -pl computer-test -am \
  -Dtest=QueuedMessageSenderTest,IntegrateTestSuite \
  -DfailIfNoTests=false -Djacoco.skip=true

Verifying these changes

• Trivial rework / code cleanup without any test coverage. (No Need)
• Already covered by existing tests, such as (please modify tests here).
• Need tests and can be verified as follows.

Does this PR potentially affect the following parts?

• Nope
• Dependencies (add/update license info)
• Modify configurations
• The public API
• Other affects (typed here)

Documentation Status

• Doc - TODO
• Doc - Done
• Doc - No Need

[Copilot]

Pull request overview

This PR addresses a control-message future race in the computer-core message-sending pipeline (specifically QueuedMessageSender) that could surface as The origin future must be null, and also tightens computer-test integration behavior to fail faster instead of stalling for long BSP timeouts.

Changes:

• Refactors control-message handling in QueuedMessageSender so each START/FINISH message carries its own CompletableFuture, and the in-flight control future is cleared before completion.
• Adds unit regressions for consecutive control messages and for transport failures completing the control future exceptionally.
• Updates sender integration tests to apply shorter BSP wait timeouts and to use a bounded waitForServices() helper.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.

File	Description
computer/computer-test/src/main/java/org/apache/hugegraph/computer/suite/integrate/SenderIntegrateTest.java	Adds CI-oriented BSP/service timeouts and a fail-fast wait helper for master/worker futures
computer/computer-test/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSenderTest.java	Adds regressions for control-future sequencing and transport exception completion
computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSender.java	Refactors control-message future lifecycle and transport-exception handling
computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessage.java	Extends queued message model to optionally carry a control future

Comments suppressed due to low confidence (2)

computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSender.java:273

• Same issue as sendStartMessage(): if setControlFuture() throws ComputerException, it will currently bubble out of sendFinishMessage() and can terminate the send-executor thread. Since the future is already completed exceptionally in setControlFuture(), catch ComputerException here and return to avoid killing the sender thread.

        public void sendFinishMessage(CompletableFuture<Void> future)
                                      throws TransportException {
            this.setControlFuture(future);
            try {
                this.client.finishSessionAsync().whenComplete((r, e) -> {

computer/computer-test/src/main/java/org/apache/hugegraph/computer/suite/integrate/SenderIntegrateTest.java:122

• The master options set withRpcServerPort() twice (8611 then 0). The first value is immediately overridden and can be misleading when debugging port binding issues; it’s clearer to keep only the effective port setting.

                                          .withRpcServerHost("127.0.0.1")
                                          .withRpcServerPort(8611)
                                          .withRpcServerPort(0)

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[imbajin]

Blocking: yes. Summary: An existing review thread covers the sender-executor termination risk, and the new fail-fast service wait can race past lifecycle cleanup. Evidence: exact-head static inspection by six independent lanes; git diff --check passed; visible exact-head checks are green.

[imbajin]

⚠️ Completing result on the first failure lets the caller enter cleanup before the other service threads have initialized and published their service references. A thread can add its service after closeWorker()/closeMaster() has already observed null or finished iterating workerServices; the multi-worker threads also do not own their services with try-with-resources. That can leave Netty/etcd resources running into later tests. Please cancel and join every spawned thread and ensure each thread closes any service it initializes in a finally, including services published after the first failure.