This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new b15b655e88 fix(admin): make upstream writes transactional (#7249)
b15b655e88 is described below
commit b15b655e8809cd76c0acc64c5037e29731518f5c
Author: Liming Deng <[email protected]>
AuthorDate: Wed Sep 30 09:47:36 2026 +0800
fix(admin): make upstream writes transactional (#7249)
---
.../controller/DiscoveryUpstreamController.java | 7 +-
.../admin/service/DiscoveryUpstreamService.java | 7 ++
.../service/impl/DiscoveryUpstreamServiceImpl.java | 21 ++++
.../DiscoveryUpstreamControllerTest.java | 8 ++
.../service/UpstreamWriteIntegrationTest.java | 134 +++++++++++++++++++++
5 files changed, 171 insertions(+), 6 deletions(-)
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/controller/DiscoveryUpstreamController.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/controller/DiscoveryUpstreamController.java
index 78a0dd4fd7..c8a5ca9267 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/controller/DiscoveryUpstreamController.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/controller/DiscoveryUpstreamController.java
@@ -17,7 +17,6 @@
package org.apache.shenyu.admin.controller;
-import org.apache.commons.collections4.CollectionUtils;
import org.apache.shenyu.admin.aspect.annotation.RestApi;
import org.apache.shenyu.admin.mapper.DiscoveryHandlerMapper;
import org.apache.shenyu.admin.model.dto.DiscoveryUpstreamDTO;
@@ -65,11 +64,7 @@ public class DiscoveryUpstreamController {
*/
@PostMapping("batch")
public ShenyuAdminResult createDiscoveryUpstreamList(@Valid @RequestBody
final List<DiscoveryUpstreamDTO> discoveryUpstreamDTOList) {
- if (CollectionUtils.isNotEmpty(discoveryUpstreamDTOList)) {
- for (DiscoveryUpstreamDTO discoveryUpstreamDTO :
discoveryUpstreamDTOList) {
- discoveryUpstreamService.createOrUpdate(discoveryUpstreamDTO);
- }
- }
+ discoveryUpstreamService.createOrUpdateBatch(discoveryUpstreamDTOList);
return ShenyuAdminResult.success();
}
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/DiscoveryUpstreamService.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/DiscoveryUpstreamService.java
index d808c194b6..047cd1821a 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/DiscoveryUpstreamService.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/DiscoveryUpstreamService.java
@@ -36,6 +36,13 @@ public interface DiscoveryUpstreamService {
*/
String createOrUpdate(DiscoveryUpstreamDTO discoveryUpstreamDTO);
+ /**
+ * Create or update upstreams in one transaction.
+ *
+ * @param upstreams upstreams to create or update
+ */
+ void createOrUpdateBatch(List<DiscoveryUpstreamDTO> upstreams);
+
/**
* updateBatch.
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/DiscoveryUpstreamServiceImpl.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/DiscoveryUpstreamServiceImpl.java
index 20cf8f8ff0..24675f44a4 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/DiscoveryUpstreamServiceImpl.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/DiscoveryUpstreamServiceImpl.java
@@ -49,6 +49,7 @@ import
org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;
import java.util.Collections;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -99,12 +100,32 @@ public class DiscoveryUpstreamServiceImpl implements
DiscoveryUpstreamService {
* @return the string
*/
@Override
+ @Transactional(rollbackFor = Exception.class)
public String createOrUpdate(final DiscoveryUpstreamDTO
discoveryUpstreamDTO) {
return StringUtils.hasLength(discoveryUpstreamDTO.getId())
? update(discoveryUpstreamDTO) : create(discoveryUpstreamDTO);
}
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public void createOrUpdateBatch(final List<DiscoveryUpstreamDTO>
upstreams) {
+ if (CollectionUtils.isEmpty(upstreams)) {
+ return;
+ }
+ Set<String> handlerIds = new LinkedHashSet<>();
+ for (DiscoveryUpstreamDTO upstream : upstreams) {
+ DiscoveryUpstreamDO entity =
DiscoveryUpstreamDO.buildDiscoveryUpstreamDO(upstream);
+ if (StringUtils.hasLength(upstream.getId())) {
+ discoveryUpstreamMapper.update(entity);
+ } else {
+ discoveryUpstreamMapper.insert(entity);
+ }
+ handlerIds.add(upstream.getDiscoveryHandlerId());
+ }
+ handlerIds.forEach(this::fetchAll);
+ }
+
@Override
@Transactional
public int updateBatch(final String discoveryHandlerId, final
List<DiscoveryUpstreamDTO> discoveryUpstreamDTOList) {
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/controller/DiscoveryUpstreamControllerTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/controller/DiscoveryUpstreamControllerTest.java
index 847ae23a90..d31658aae5 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/controller/DiscoveryUpstreamControllerTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/controller/DiscoveryUpstreamControllerTest.java
@@ -46,6 +46,7 @@ import static
org.apache.shenyu.common.constant.Constants.SYS_DEFAULT_NAMESPACE_
import static org.hamcrest.Matchers.containsString;
import static org.mockito.BDDMockito.given;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
@@ -133,6 +134,13 @@ public final class DiscoveryUpstreamControllerTest {
.andExpect(jsonPath("$.code").value(200));
}
+ @Test
+ public void testBatchDelegatesToTransactionalService() {
+ List<DiscoveryUpstreamDTO> upstreams = List.of(buildDTO("handler"));
+ discoveryUpstreamController.createDiscoveryUpstreamList(upstreams);
+ verify(discoveryUpstreamService).createOrUpdateBatch(upstreams);
+ }
+
private DiscoveryUpstreamDTO buildDTO(final String discoveryHandlerId) {
DiscoveryUpstreamDTO dto = new DiscoveryUpstreamDTO();
dto.setDiscoveryHandlerId(discoveryHandlerId);
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/UpstreamWriteIntegrationTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/UpstreamWriteIntegrationTest.java
new file mode 100644
index 0000000000..34995a14a9
--- /dev/null
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/UpstreamWriteIntegrationTest.java
@@ -0,0 +1,134 @@
+/*
+ * 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.shenyu.admin.service;
+
+import jakarta.annotation.Resource;
+import org.apache.shenyu.admin.AbstractSpringIntegrationTest;
+import org.apache.shenyu.admin.discovery.DiscoveryProcessor;
+import org.apache.shenyu.admin.discovery.DiscoveryProcessorHolder;
+import org.apache.shenyu.admin.model.dto.DiscoveryUpstreamDTO;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+/**
+ * Transaction regression tests for upstream writes.
+ */
+public class UpstreamWriteIntegrationTest extends
AbstractSpringIntegrationTest {
+
+ @Resource
+ private DiscoveryUpstreamService upstreamService;
+
+ @Resource
+ private JdbcTemplate jdbcTemplate;
+
+ @MockBean
+ private DiscoveryProcessorHolder processorHolder;
+
+ private DiscoveryProcessor processor;
+
+ @BeforeEach
+ public void setup() {
+ processor = mock(DiscoveryProcessor.class);
+ when(processorHolder.chooseProcessor("local")).thenReturn(processor);
+ jdbcTemplate.update("INSERT INTO discovery (id, discovery_name,
discovery_level, namespace_id, discovery_type) "
+ + "VALUES ('write-test', 'write-test', '0', 'write-ns',
'local')");
+ jdbcTemplate.update("INSERT INTO discovery_handler (id, discovery_id,
handler) VALUES ('write-test', 'write-test', '{}')");
+ jdbcTemplate.update("INSERT INTO proxy_selector (id, name,
plugin_name, type, forward_port, namespace_id) "
+ + "VALUES ('write-test', 'write-test', 'tcp', 'tcp', 18080,
'write-ns')");
+ jdbcTemplate.update("INSERT INTO discovery_rel (id, plugin_name,
discovery_handler_id, proxy_selector_id) "
+ + "VALUES ('write-test', 'tcp', 'write-test', 'write-test')");
+ }
+
+ @AfterEach
+ public void cleanup() {
+ jdbcTemplate.update("DELETE FROM discovery_upstream WHERE
discovery_handler_id = 'write-test'");
+ jdbcTemplate.update("DELETE FROM discovery_rel WHERE id =
'write-test'");
+ jdbcTemplate.update("DELETE FROM proxy_selector WHERE id =
'write-test'");
+ jdbcTemplate.update("DELETE FROM discovery_handler WHERE id =
'write-test'");
+ jdbcTemplate.update("DELETE FROM discovery WHERE id = 'write-test'");
+ }
+
+ @Test
+ public void testCreateRollsBackOnSyncFailure() {
+ failSync();
+ assertThrows(IllegalStateException.class, () ->
upstreamService.createOrUpdate(upstream("localhost:8080")));
+ assertEquals(0, countUpstreams());
+ }
+
+ @Test
+ public void testUpdateRollsBackOnSyncFailure() {
+ upstreamService.createOrUpdate(upstream("localhost:8080"));
+ DiscoveryUpstreamDTO update = upstream("localhost:8081");
+ update.setId(jdbcTemplate.queryForObject("SELECT id FROM
discovery_upstream WHERE discovery_handler_id = 'write-test'", String.class));
+ failSync();
+ assertThrows(IllegalStateException.class, () ->
upstreamService.createOrUpdate(update));
+ assertEquals("localhost:8080", jdbcTemplate.queryForObject("SELECT
upstream_url FROM discovery_upstream WHERE id = ?", String.class,
update.getId()));
+ }
+
+ @Test
+ public void testBatchInsertFailureRollsBackBeforeSync() {
+ assertThrows(DataIntegrityViolationException.class,
+ () ->
upstreamService.createOrUpdateBatch(List.of(upstream("localhost:8080"),
upstream(null))));
+ assertEquals(0, countUpstreams());
+ verifyNoInteractions(processor);
+ }
+
+ @Test
+ public void testBatchSyncFailureRollsBackAllRows() {
+ failSync();
+ assertThrows(IllegalStateException.class,
+ () ->
upstreamService.createOrUpdateBatch(List.of(upstream("localhost:8080"),
upstream("localhost:8081"))));
+ assertEquals(0, countUpstreams());
+ }
+
+ private void failSync() {
+ doThrow(new IllegalStateException("sync
failed")).when(processor).changeUpstream(any(), any());
+ }
+
+ private int countUpstreams() {
+ return jdbcTemplate.queryForObject("SELECT COUNT(*) FROM
discovery_upstream WHERE discovery_handler_id = 'write-test'", Integer.class);
+ }
+
+ private DiscoveryUpstreamDTO upstream(final String url) {
+ DiscoveryUpstreamDTO dto = new DiscoveryUpstreamDTO();
+ dto.setDiscoveryHandlerId("write-test");
+ dto.setNamespaceId("write-ns");
+ dto.setProtocol("tcp");
+ dto.setUrl(url);
+ dto.setStatus(0);
+ dto.setWeight(50);
+ dto.setProps("{}");
+ return dto;
+ }
+}
+
