fix: fix derby data import in standalone mode

client/src/test/java/com/alibaba/nacos/client/NamingTest.java

@@ -62,8 +62,7 @@ public void testServiceList() throws Exception {
 
         System.out.println(list);
 
-        CountDownLatch latch = new CountDownLatch(1);
-        latch.await();
+        ThreadUtils.sleep(30_000L);
 //        ExpressionSelector expressionSelector = new ExpressionSelector();
 //        expressionSelector.setExpression("INSTANCE.metadata.registerSource = 'dubbo'");
 //        ListView<String> serviceList = namingService.getServicesOfServer(1, 10, expressionSelector);

config/src/main/java/com/alibaba/nacos/config/server/controller/ConfigOpsController.java

@@ -19,7 +19,7 @@
 import com.alibaba.nacos.common.model.RestResultUtils;
 import com.alibaba.nacos.common.utils.Objects;
 import com.alibaba.nacos.config.server.constant.Constants;
-import com.alibaba.nacos.config.server.model.event.DataImportEvent;
+import com.alibaba.nacos.config.server.model.event.DerbyImportEvent;
 import com.alibaba.nacos.config.server.service.datasource.DynamicDataSource;
 import com.alibaba.nacos.config.server.service.datasource.LocalDataSourceServiceImpl;
 import com.alibaba.nacos.config.server.service.repository.PersistService;
@@ -134,10 +134,10 @@ public DeferredResult<RestResult<String>> importDerby(
 			return response;
 		}
 		DatabaseOperate databaseOperate = ApplicationUtils.getBean(DatabaseOperate.class);
-		WebUtils.onFileUpload(multipartFile, path -> {
-			NotifyCenter.publishEvent(new DataImportEvent(false));
-			databaseOperate.dataImport(path.toFile()).whenComplete((result, ex) -> {
-				NotifyCenter.publishEvent(new DataImportEvent(true));
+		WebUtils.onFileUpload(multipartFile, file -> {
+			NotifyCenter.publishEvent(new DerbyImportEvent(false));
+			databaseOperate.dataImport(file).whenComplete((result, ex) -> {
+				NotifyCenter.publishEvent(new DerbyImportEvent(true));
 				if (Objects.nonNull(ex)) {
 					response.setResult(RestResultUtils.failed(ex.getMessage()));
 					return;

config/src/main/java/com/alibaba/nacos/config/server/filter/CurcuitFilter.java

@@ -18,7 +18,6 @@
 
 import com.alibaba.nacos.common.utils.ExceptionUtil;
 import com.alibaba.nacos.config.server.constant.Constants;
-import com.alibaba.nacos.config.server.model.event.DataImportEvent;
 import com.alibaba.nacos.config.server.model.event.RaftDBErrorEvent;
 import com.alibaba.nacos.config.server.model.event.RaftDBErrorRecoverEvent;
 import com.alibaba.nacos.consistency.cp.CPProtocol;
@@ -137,17 +136,12 @@ public void onEvent(Event event) {
 				if (event instanceof RaftDBErrorEvent) {
 					isDowngrading = true;
 				}
-				if (event instanceof DataImportEvent) {
-					DataImportEvent e = (DataImportEvent) event;
-					isOpenService = e.isFinished();
-				}
 			}
 
 			@Override
 			public boolean canNotify(Event event) {
 				return (event instanceof RaftDBErrorEvent)
-						|| (event instanceof RaftDBErrorRecoverEvent)
-						|| (event instanceof DataImportEvent);
+						|| (event instanceof RaftDBErrorRecoverEvent);
 			}
 		});
 	}

config/src/main/java/com/alibaba/nacos/config/server/model/event/DataImportEvent.java → config/src/main/java/com/alibaba/nacos/config/server/model/event/DerbyImportEvent.java

@@ -23,13 +23,13 @@
  *
  * @author <a href="mailto:liaochuntao@live.com">liaochuntao</a>
  */
-public class DataImportEvent implements SlowEvent {
+public class DerbyImportEvent implements SlowEvent {
 
 	private static final long serialVersionUID = 3299565864352399053L;
 
 	private final boolean finished;
 
-	public DataImportEvent(boolean finished) {
+	public DerbyImportEvent(boolean finished) {
 		this.finished = finished;
 	}
 

config/src/main/java/com/alibaba/nacos/config/server/service/repository/embedded/DistributedDatabaseOperateImpl.java

@@ -425,7 +425,7 @@ public CompletableFuture<RestResult<String>> dataImport(File file) {
 						batchUpdate.clear();
 					}
 				}
-				CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
+				CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
 				for (CompletableFuture<Response> future : futures) {
 					Response response = future.get();
 					if (!response.getSuccess()) {

config/src/main/java/com/alibaba/nacos/config/server/service/repository/embedded/EmbeddedStoragePersistServiceImpl.java

@@ -38,7 +38,7 @@
 import com.alibaba.nacos.config.server.model.SameConfigPolicy;
 import com.alibaba.nacos.config.server.model.SubInfo;
 import com.alibaba.nacos.config.server.model.TenantInfo;
-import com.alibaba.nacos.config.server.model.event.DataImportEvent;
+import com.alibaba.nacos.config.server.model.event.DerbyImportEvent;
 import com.alibaba.nacos.config.server.service.datasource.DataSourceService;
 import com.alibaba.nacos.config.server.service.datasource.DynamicDataSource;
 import com.alibaba.nacos.config.server.service.repository.PaginationHelper;
@@ -124,7 +124,7 @@ public EmbeddedStoragePersistServiceImpl(DatabaseOperate databaseOperate,
 		this.databaseOperate = databaseOperate;
 		this.idGeneratorManager = idGeneratorManager;
 
-		NotifyCenter.registerToSharePublisher(DataImportEvent.class);
+		NotifyCenter.registerToSharePublisher(DerbyImportEvent.class);
 	}
 
 	@PostConstruct

config/src/main/java/com/alibaba/nacos/config/server/service/repository/embedded/StandaloneDatabaseOperateImpl.java

@@ -34,7 +34,6 @@
 
 import javax.annotation.PostConstruct;
 import java.io.File;
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -44,99 +43,96 @@
 import java.util.stream.Collectors;
 
 /**
+ * Derby operation in stand-alone mode.
+ *
  * @author <a href="mailto:liaochuntao@live.com">liaochuntao</a>
  */
 @Conditional(ConditionStandaloneEmbedStorage.class)
 @Component
 public class StandaloneDatabaseOperateImpl implements BaseDatabaseOperate {
-
-	private DataSourceService dataSourceService;
-
-	private JdbcTemplate jdbcTemplate;
-	private TransactionTemplate transactionTemplate;
-
-	@PostConstruct
-	protected void init() {
-		dataSourceService = DynamicDataSource.getInstance().getDataSource();
-		jdbcTemplate = dataSourceService.getJdbcTemplate();
-		transactionTemplate = dataSourceService.getTransactionTemplate();
-		LogUtil.defaultLog.info("use StandaloneDatabaseOperateImpl");
-	}
-
-	@Override
-	public <R> R queryOne(String sql, Class<R> cls) {
-		return queryOne(jdbcTemplate, sql, cls);
-	}
-
-	@Override
-	public <R> R queryOne(String sql, Object[] args, Class<R> cls) {
-		return queryOne(jdbcTemplate, sql, args, cls);
-	}
-
-	@Override
-	public <R> R queryOne(String sql, Object[] args, RowMapper<R> mapper) {
-		return queryOne(jdbcTemplate, sql, args, mapper);
-	}
-
-	@Override
-	public <R> List<R> queryMany(String sql, Object[] args, RowMapper<R> mapper) {
-		return queryMany(jdbcTemplate, sql, args, mapper);
-	}
-
-	@Override
-	public <R> List<R> queryMany(String sql, Object[] args, Class<R> rClass) {
-		return queryMany(jdbcTemplate, sql, args, rClass);
-	}
-
-	@Override
-	public List<Map<String, Object>> queryMany(String sql, Object[] args) {
-		return queryMany(jdbcTemplate, sql, args);
-	}
-
-	@Override
-	public CompletableFuture<RestResult<String>> dataImport(File file) {
-		return CompletableFuture.supplyAsync(() -> {
-			try (DiskUtils.LineIterator iterator = DiskUtils.lineIterator(file)) {
-				int batchSize = 1000;
-				List<String> batchUpdate = new ArrayList<>(batchSize);
-				List<CompletableFuture<Void>> futures = new ArrayList<>();
-				List<Boolean> results = new CopyOnWriteArrayList<>();
-				while (iterator.hasNext()) {
-					String sql = iterator.next();
-					if (StringUtils.isBlank(sql)) {
-						continue;
-					}
-					batchUpdate.add(sql);
-					if (batchUpdate.size() == batchSize ||!iterator.hasNext()) {
-						futures.add(CompletableFuture.runAsync(
-								() -> results.add(doDataImport(jdbcTemplate, batchUpdate.stream()
-										.map(s -> {
-											ModifyRequest request = new ModifyRequest();
-											request.setSql(s);
-											return request;
-										}).collect(Collectors.toList())))));
-						batchUpdate.clear();
-					}
-				}
-				CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
-				return RestResult.<String>builder()
-						.withCode(BooleanUtils.and(results.toArray(new Boolean[0])) ? 200 : 500)
-						.withData("").build();
-			} catch (Throwable ex) {
-				LogUtil.defaultLog.error("data import has error : {}", ex);
-				return RestResultUtils.failed(ex.getMessage());
-			}
-		});
-	}
-
-	@Override
-	public Boolean update(List<ModifyRequest> modifyRequests,
-			BiConsumer<Boolean, Throwable> consumer) {
-		return update(modifyRequests);
-	}
-
-	@Override
-	public Boolean update(List<ModifyRequest> requestList) {
-		return update(transactionTemplate, jdbcTemplate, requestList);
-	}
+
+    private JdbcTemplate jdbcTemplate;
+
+    private TransactionTemplate transactionTemplate;
+
+    @PostConstruct
+    protected void init() {
+        DataSourceService dataSourceService = DynamicDataSource.getInstance().getDataSource();
+        jdbcTemplate = dataSourceService.getJdbcTemplate();
+        transactionTemplate = dataSourceService.getTransactionTemplate();
+        LogUtil.defaultLog.info("use StandaloneDatabaseOperateImpl");
+    }
+
+    @Override
+    public <R> R queryOne(String sql, Class<R> cls) {
+        return queryOne(jdbcTemplate, sql, cls);
+    }
+
+    @Override
+    public <R> R queryOne(String sql, Object[] args, Class<R> cls) {
+        return queryOne(jdbcTemplate, sql, args, cls);
+    }
+
+    @Override
+    public <R> R queryOne(String sql, Object[] args, RowMapper<R> mapper) {
+        return queryOne(jdbcTemplate, sql, args, mapper);
+    }
+
+    @Override
+    public <R> List<R> queryMany(String sql, Object[] args, RowMapper<R> mapper) {
+        return queryMany(jdbcTemplate, sql, args, mapper);
+    }
+
+    @Override
+    public <R> List<R> queryMany(String sql, Object[] args, Class<R> rClass) {
+        return queryMany(jdbcTemplate, sql, args, rClass);
+    }
+
+    @Override
+    public List<Map<String, Object>> queryMany(String sql, Object[] args) {
+        return queryMany(jdbcTemplate, sql, args);
+    }
+
+    @Override
+    public CompletableFuture<RestResult<String>> dataImport(File file) {
+        return CompletableFuture.supplyAsync(() -> {
+            try (DiskUtils.LineIterator iterator = DiskUtils.lineIterator(file)) {
+                int batchSize = 1000;
+                List<String> batchUpdate = new ArrayList<>(batchSize);
+                List<CompletableFuture<Void>> futures = new ArrayList<>();
+                List<Boolean> results = new CopyOnWriteArrayList<>();
+                while (iterator.hasNext()) {
+                    String sql = iterator.next();
+                    if (StringUtils.isNotBlank(sql)) {
+                        batchUpdate.add(sql);
+                    }
+                    if (batchUpdate.size() == batchSize || !iterator.hasNext()) {
+                        List<ModifyRequest> sqls = batchUpdate.stream().map(s -> {
+                            ModifyRequest request = new ModifyRequest();
+                            request.setSql(s);
+                            return request;
+                        }).collect(Collectors.toList());
+                        futures.add(CompletableFuture.runAsync(() -> results.add(doDataImport(jdbcTemplate, sqls))));
+                        batchUpdate.clear();
+                    }
+                }
+                CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+                return RestResult.<String>builder()
+                        .withCode(BooleanUtils.and(results.toArray(new Boolean[0])) ? 200 : 500).withData("").build();
+            } catch (Throwable ex) {
+                LogUtil.defaultLog.error("An exception occurred when external data was imported into Derby : {}", ex);
+                return RestResultUtils.failed(ex.getMessage());
+            }
+        });
+    }
+
+    @Override
+    public Boolean update(List<ModifyRequest> modifyRequests, BiConsumer<Boolean, Throwable> consumer) {
+        return update(modifyRequests);
+    }
+
+    @Override
+    public Boolean update(List<ModifyRequest> requestList) {
+        return update(transactionTemplate, jdbcTemplate, requestList);
+    }
 }

core/src/main/java/com/alibaba/nacos/core/notify/DefaultPublisher.java

@@ -27,6 +27,7 @@
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 
 import static com.alibaba.nacos.core.notify.NotifyCenter.RING_BUFFER_SIZE;
@@ -95,13 +96,15 @@ public void run() {
 
     void openEventHandler() {
         try {
+            int waitTimes = 60;
             // To ensure that messages are not lost, enable EventHandler when
             // waiting for the first Subscriber to register
             for (; ; ) {
-                if (shutdown || canOpen) {
+                if (shutdown || canOpen || waitTimes <= 0) {
                     break;
                 }
                 ThreadUtils.sleep(1_000L);
+                waitTimes--;
             }
 
             for (; ; ) {