You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "youngkermit8-coder (via GitHub)" <gi...@apache.org> on 2026/10/20 02:23:52 UTC
[I] [Bug] Topic and consumer-group batch imports submit later requests after interruption [rocketmq-dashboard]
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: commits-unsubscribe@rocketmq.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org