youngkermit8-coder opened a new issue, #7197:
URL: https://github.com/apache/rocketmq-dashboard/issues/7197
### Before Creating the Bug Report
- [x] I searched all-state issues and PRs for import/interruption,
importTopics/cancel, importConsumerGroups/interrupt and batch-import
cancellation; the finite searches did not identify an equivalent outer-batch
fix.
- [x] This concerns Studio configuration-import orchestration, not a broker
usage question.
- [x] The exact source version and controlled reproduction are stated below.
### Studio Version
Source-built `rocketmq-studio@127c47b06a268cc74224d5f34c56ec7c36ac0d55`.
This was still the branch tip during the current reproduction/preflight. No
production fix, POM change or dependency override was applied; only the test
below was added to an isolated source export.
### Runtime Environment
Windows, Temurin 21.0.12+8, Maven 3.9.9. The real MetadataService,
AliyunInstanceProvider and AliyunClientFactory run with the official
AsyncClient mocked at the network boundary. Instance routing, ownership guard
and audit collaborators are mocked. No HTTP application, real cloud, database
or broker is started. The fixture uses synthetic resource identifiers and never
reads credentials.
### Connected RocketMQ Cluster
None. Recorded SDK invocations are mock submissions, not confirmed cloud
resource creation or a production incident.
### Build Toolchain
JDK21 / Maven3.9.9. The current targeted test run intentionally exits1 with
eight assertions showing unwanted later SDK requests; it is not a successful
fixed regression, full suite or package run. Checkstyle reports zero
violations. Existing effective-model/annotation-processing/deprecation warnings
remain in the local log.
### Describe the Bug
`MetadataService.importTopics` and `importConsumerGroups` catch each row's
exception and proceed to the next row. If the real AliyunClientFactory is
interrupted while waiting for a pending SDK Future, it cancels that Future and
restores the interrupt flag. The import loop then still invokes the next row's
SDK creation method. A completed Future can return successfully on an
already-interrupted thread, so another client-factory wait does not reliably
stop the batch.
This is a cooperative row-boundary admission gap. It does not mean that
closing a browser necessarily interrupts a server thread, or that an
already-issued request can be rolled back.
### Steps to Reproduce
At the pinned base, add the following exact fixture at
`server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataImportInterruptionTest.java`:
```java
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.studio.instance.topic;
import com.aliyun.sdk.service.rocketmq20220801.AsyncClient;
import
com.aliyun.sdk.service.rocketmq20220801.models.CreateConsumerGroupRequest;
import
com.aliyun.sdk.service.rocketmq20220801.models.CreateConsumerGroupResponse;
import com.aliyun.sdk.service.rocketmq20220801.models.CreateTopicRequest;
import com.aliyun.sdk.service.rocketmq20220801.models.CreateTopicResponse;
import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.domain.enums.TopicType;
import org.apache.rocketmq.studio.instance.InstanceRepository;
import org.apache.rocketmq.studio.instance.InstanceResolver;
import org.apache.rocketmq.studio.instance.InstanceVO;
import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard;
import org.apache.rocketmq.studio.instance.group.CreateConsumerGroupDTO;
import org.apache.rocketmq.studio.instance.message.MessageService;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import org.apache.rocketmq.studio.provider.alibaba.AliyunClientFactory;
import org.apache.rocketmq.studio.provider.alibaba.AliyunInstanceProvider;
import org.apache.rocketmq.studio.provider.apache.AdminClient;
import org.apache.rocketmq.studio.provider.apache.MetadataProvider;
import
org.apache.rocketmq.studio.provider.credential.CloudCredentialRepository;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/** Exercises real service, provider and future-interruption handling with a
network-free SDK boundary. */
@Timeout(10)
class MetadataImportInterruptionTest {
private final AsyncClient sdk = mock(AsyncClient.class);
private final List<String> submitted = new ArrayList<>();
private final AliyunClientFactory factory = new
AliyunClientFactory(mock(CloudCredentialRepository.class)) {
@Override
public AsyncClient client(Long credentialId, String region) {
return sdk;
}
};
private MetadataService service;
private String mode = "success";
private CompletableFuture<?> interruptedFuture;
private final CountDownLatch pendingRequest = new CountDownLatch(1);
@BeforeEach
void setUpTest() {
InstanceVO instance = InstanceVO.builder().name("local-fixture")
.vendor(InstanceVendor.ALIYUN).cloudInstanceId("synthetic-cloud-id")
.regionId("cn-hangzhou").credentialId(1L).build();
InstanceRepository repository = mock(InstanceRepository.class);
when(repository.findByIdentifier("local-fixture")).thenReturn(Optional.of(instance));
AliyunInstanceProvider provider = new
AliyunInstanceProvider(factory, repository);
InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
when(registry.byInstanceId("local-fixture")).thenReturn(Optional.of(provider));
InstanceResolver resolver = mock(InstanceResolver.class);
when(resolver.findByIdentifier("local-fixture")).thenReturn(Optional.of(instance));
ResourceOwnershipGuard guard = mock(ResourceOwnershipGuard.class);
when(guard.requireInstance("local-fixture")).thenReturn(instance);
service = new MetadataService(mock(MetadataProvider.class),
mock(AdminClient.class), registry, resolver,
mock(OperationAuditService.class),
mock(MessageService.class),
mock(RuntimeAdminClientResolver.class), guard);
when(sdk.createTopic(any(CreateTopicRequest.class))).thenAnswer(call
-> {
CreateTopicRequest request = call.getArgument(0);
return complete(request.getTopicName(),
CreateTopicResponse.create().toBuilder().statusCode(200).build());
});
when(sdk.createConsumerGroup(any(CreateConsumerGroupRequest.class))).thenAnswer(call
-> {
CreateConsumerGroupRequest request = call.getArgument(0);
return complete(request.getConsumerGroupId(),
CreateConsumerGroupResponse.create().toBuilder().statusCode(200).build());
});
}
private <T> CompletableFuture<T> complete(String name, T response) {
submitted.add(name);
if ("row-1".equals(name)) {
if ("interrupt".equals(mode) ||
"external-interrupt".equals(mode)) {
CompletableFuture<T> pending = new CompletableFuture<>();
interruptedFuture = pending;
// The actual factory's timed get must cancel this future
and restore the flag.
if ("interrupt".equals(mode)) {
Thread.currentThread().interrupt();
}
pendingRequest.countDown();
return pending;
}
if ("failure".equals(mode)) {
return CompletableFuture.failedFuture(new
IllegalStateException("ordinary SDK failure"));
}
if ("success-interrupted".equals(mode)) {
Thread.currentThread().interrupt();
}
}
return CompletableFuture.completedFuture(response);
}
@AfterEach
void clearFixtureInterruptTest() {
Thread.interrupted();
factory.close();
}
@ParameterizedTest
@ValueSource(booleans = {false, true})
void interruptionStopsLaterSdkRequestsTest(boolean groups) {
mode = "interrupt";
Outcome result = importRows(groups);
assertThat(interruptedFuture).isNotNull();
assertThat(interruptedFuture.isCancelled()).isTrue();
assertThat(Thread.currentThread().isInterrupted()).isTrue();
assertThat(submitted).containsExactly("row-0", "row-1");
assertThat(result.imported()).isEqualTo(1);
assertThat(result.failed()).isEqualTo(2);
assertThat(result.failureIndices()).containsExactly(1, 2);
assertThat(result.messages().getLast()).contains("not attempted");
}
@ParameterizedTest
@ValueSource(booleans = {false, true})
void
anotherThreadCanInterruptPendingBatchWithoutFurtherWritesTest(boolean groups)
throws Exception {
mode = "external-interrupt";
AtomicBoolean interruptRetained = new AtomicBoolean();
FutureTask<Outcome> task = new FutureTask<>(() -> {
Outcome result = importRows(groups);
interruptRetained.set(Thread.currentThread().isInterrupted());
return result;
});
Thread worker = new Thread(task,
"metadata-import-interruption-fixture");
worker.start();
try {
assertThat(pendingRequest.await(5, TimeUnit.SECONDS)).isTrue();
worker.interrupt();
Outcome result = task.get(5, TimeUnit.SECONDS);
assertThat(interruptedFuture.isCancelled()).isTrue();
assertThat(interruptRetained).isTrue();
assertThat(submitted).containsExactly("row-0", "row-1");
assertThat(result.imported()).isEqualTo(1);
assertThat(result.failed()).isEqualTo(2);
assertThat(result.failureIndices()).containsExactly(1, 2);
} finally {
worker.interrupt();
worker.join(2_000);
assertThat(worker.isAlive()).isFalse();
}
}
@ParameterizedTest
@ValueSource(booleans = {false, true})
void completedRowIsRetainedWhenInterruptArrivesBeforeNextRowTest(boolean
groups) {
mode = "success-interrupted";
Outcome result = importRows(groups);
assertThat(Thread.currentThread().isInterrupted()).isTrue();
assertThat(submitted).containsExactly("row-0", "row-1");
assertThat(result.imported()).isEqualTo(2);
assertThat(result.failed()).isEqualTo(1);
assertThat(result.failureIndices()).containsExactly(2);
}
@ParameterizedTest
@ValueSource(booleans = {false, true})
void alreadyInterruptedBatchDoesNotSubmitSdkRequestsTest(boolean groups)
{
Thread.currentThread().interrupt();
Outcome result = importRows(groups);
assertThat(Thread.currentThread().isInterrupted()).isTrue();
assertThat(submitted).isEmpty();
assertThat(result.imported()).isZero();
assertThat(result.failed()).isEqualTo(3);
assertThat(result.failureIndices()).containsExactly(0, 1, 2);
}
@ParameterizedTest
@ValueSource(booleans = {false, true})
void ordinarySdkFailureStillAllowsOtherRowsTest(boolean groups) {
mode = "failure";
Outcome result = importRows(groups);
assertThat(Thread.currentThread().isInterrupted()).isFalse();
assertThat(submitted).containsExactly("row-0", "row-1", "row-2");
assertThat(result.imported()).isEqualTo(2);
assertThat(result.failed()).isEqualTo(1);
assertThat(result.failureIndices()).containsExactly(1);
}
@ParameterizedTest
@ValueSource(booleans = {false, true})
void uninterruptedBatchImportsAllRowsTest(boolean groups) {
Outcome result = importRows(groups);
assertThat(submitted).containsExactly("row-0", "row-1", "row-2");
assertThat(result.imported()).isEqualTo(3);
assertThat(result.failed()).isZero();
}
private Outcome importRows(boolean groups) {
List<String> names = List.of("row-0", "row-1", "row-2");
if (groups) {
var result = service.importConsumerGroups("local-fixture",
names.stream().map(name -> {
CreateConsumerGroupDTO dto = new CreateConsumerGroupDTO();
dto.setName(name);
return dto;
}).toList());
return new Outcome(result.getImported(), result.getFailed(),
result.getFailures().stream().map(failure ->
failure.getIndex()).toList(),
result.getFailures().stream().map(failure ->
failure.getMessage()).toList());
}
var result = service.importTopics("local-fixture",
names.stream().map(name -> {
CreateTopicDTO dto = new CreateTopicDTO();
dto.setName(name);
dto.setType(TopicType.NORMAL);
return dto;
}).toList());
return new Outcome(result.getImported(), result.getFailed(),
result.getFailures().stream().map(failure ->
failure.getIndex()).toList(),
result.getFailures().stream().map(failure ->
failure.getMessage()).toList());
}
private record Outcome(int imported, int failed, List<Integer>
failureIndices, List<String> messages) {
}
}
```
Run from `server/` using Java21:
```sh
mvn -B -ntp -Dtest=MetadataImportInterruptionTest
-Dsurefire.failIfNoSpecifiedTests=true test
```
The current Windows invocation used offline cached dependencies, bounded
heaps and the existing Mockito agent explicitly. The external-interrupt cases
wait on a latch for row1's pending Future, then interrupt the actual importing
worker and join it with a bound; there is no network call or timing sleep.
Cleanup clears only the test thread's flag and closes the fixture factory.
### What Did You Expect to See?
Once an interrupt is visible at a row boundary, do not start provider
mutations for later rows; preserve the flag and already-confirmed successes. A
non-interruption row failure should continue to permit subsequent rows.
The representation of unattempted rows needs agreement: should the existing
`failed`/`failures` fields include them with original indices and an explicit
"not attempted" reason, or should the batch use another established response
policy? The fixture contains assertions for the first option, but the baseline
failures occur earlier at unwanted SDK submission assertions and do not
establish maintainer agreement on that response design.
### What Did You See Instead?
Twelve invocations on the unchanged current production implementation: eight
interruption assertions fail, four healthy/ordinary-failure controls pass, zero
errors or skips.
- Pending row1 interrupted in the importing thread: both import paths still
submit row2; the pending Future is cancelled and the flag is retained.
- A separate thread interrupts the pending row1: both paths still submit
row2.
- Row1 completes while the flag becomes set: both paths still submit row2
instead of stopping after confirmed success.
- The batch begins interrupted: both paths submit rows0,1,2 rather than no
SDK requests.
- With no interruption, all three rows succeed; with an ordinary exceptional
row1 Future, row2 proceeds. These four controls pass.
An independent local audit verified the 1142 original exported files against
frozen archive/disk and pinned Git blobs, allowing only the export's CRLF-to-LF
conversion. The initial auditor rejected raw-byte identity because `git
archive` inherited `core.autocrlf=true`; that failed audit is preserved, and
the follow-up proved only line-ending differences without changing source, XML
or rerunning tests. The three compiled production classes match a separately
built pure baseline at the same commit. This provenance audit is not additional
behavioral coverage or a fixed/full-build result.
### Additional Context
This differs from #1519/#1520, which address interruption of an individual
Aliyun SDK Future and are already represented in the baseline. #5544 concerns
DLQ processing, not topic/consumer-group configuration imports. Both import
methods share one root cause and belong in one contribution, not separate PRs.
Searches are bounded, not proof that no differently worded related work exists.
Earlier local prototypes and old-snapshot regressions predate this Issue.
They are not claimed as current-base implementation validation or consensus. An
older separate real-MySQL/Hikari audit probe observed that an interrupted
failure audit can be lost when the connection pool is occupied; unattempted
rows were not individually audited. The reproduction above mocks audit and does
not retest or solve that limitation. A row-boundary check is cooperative, not
atomic fencing, rollback, exactly-once execution or an audit-durability
guarantee. The already-issued interrupted row's actual cloud state must be
checked before retrying in a real deployment.
Could maintainers confirm the intended partial-result and audit semantics
before a code PR? No API/schema/dependency rewrite is proposed here. No code
PR, binaries, raw logs, database, credentials or unrelated patches are
attached. AI assistance was used for investigation and verification. No contest
score or merge outcome is claimed.
### Are You Willing to Submit a Pull Request?
- [x] Yes, after the necessary response/audit feedback, fresh implementation
regression and normal packaging checks.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
