Skip to content

Commit

Permalink
[AMORO-2217] 0.4.x fix clear and reset ServiceContainer after HA (#2221)
Browse files Browse the repository at this point in the history 
ServiceContainer supports reset before start
  • Loading branch information
@wangtaohz
wangtaohz authored Nov 3, 2023
1 parent 0472fe2 commit bb38fb5
Show file tree
Hide file tree
Showing 2 changed files with 42 additions and 23 deletions.
1 change: 1 addition & 0 deletions ams/ams-server/src/main/java/com/netease/arctic/ams/server/ArcticMetaStore.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -187,6 +187,7 @@ private static String getArcticHome() {
}

public static void startMetaStore(Configuration conf) throws Throwable {
ServiceContainer.reset();
//prepare env
if (conf.getString(ArcticMetaStoreConf.DB_TYPE).equals("derby")) {
DerbyService derbyService = new DerbyService();
Expand Down
64 changes: 41 additions & 23 deletions ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/ServiceContainer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@

package com.netease.arctic.ams.server.service;

import com.google.common.annotations.VisibleForTesting;
import com.netease.arctic.ams.server.ArcticMetaStore;
import com.netease.arctic.ams.server.handler.impl.ArcticTableMetastoreHandler;
import com.netease.arctic.ams.server.handler.impl.OptimizeManagerHandler;
Expand Down Expand Up @@ -97,11 +96,14 @@ public class ServiceContainer {

private static volatile TerminalManager terminalManager;

public static volatile PlatformFileInfoService platformFileInfoService;
private static volatile PlatformFileInfoService platformFileInfoService;

private static volatile TableBlockerService tableBlockerService;

public static volatile TableBlockerService tableBlockerService;
private static volatile boolean closed = false;

public static IOptimizeService getOptimizeService() {
checkNotClosed();
if (optimizeService == null) {
synchronized (ServiceContainer.class) {
if (optimizeService == null) {
Expand All @@ -114,6 +116,7 @@ public static IOptimizeService getOptimizeService() {
}

public static ITableExpireService getTableExpireService() {
checkNotClosed();
if (tableExpireService == null) {
synchronized (ServiceContainer.class) {
if (tableExpireService == null) {
Expand All @@ -126,6 +129,7 @@ public static ITableExpireService getTableExpireService() {
}

public static IOrphanFilesCleanService getOrphanFilesCleanService() {
checkNotClosed();
if (orphanFilesCleanService == null) {
synchronized (ServiceContainer.class) {
if (orphanFilesCleanService == null) {
Expand All @@ -138,6 +142,7 @@ public static IOrphanFilesCleanService getOrphanFilesCleanService() {
}

public static TrashCleanService getTrashCleanService() {
checkNotClosed();
if (trashCleanService == null) {
synchronized (ServiceContainer.class) {
if (trashCleanService == null) {
Expand All @@ -150,6 +155,7 @@ public static TrashCleanService getTrashCleanService() {
}

public static OptimizerService getOptimizerService() {
checkNotClosed();
if (optimizerService == null) {
synchronized (ServiceContainer.class) {
if (optimizerService == null) {
Expand All @@ -161,6 +167,7 @@ public static OptimizerService getOptimizerService() {
}

public static OptimizeManagerHandler getOptimizeManagerHandler() {
checkNotClosed();
if (optimizeManagerHandler == null) {
synchronized (ServiceContainer.class) {
if (optimizeManagerHandler == null) {
Expand All @@ -172,6 +179,7 @@ public static OptimizeManagerHandler getOptimizeManagerHandler() {
}

public static OptimizeQueueService getOptimizeQueueService() {
checkNotClosed();
if (optimizeQueueService == null) {
synchronized (ServiceContainer.class) {
if (optimizeQueueService == null) {
Expand All @@ -184,6 +192,7 @@ public static OptimizeQueueService getOptimizeQueueService() {
}

public static IMetaService getMetaService() {
checkNotClosed();
if (metaService == null) {
synchronized (ServiceContainer.class) {
if (metaService == null) {
Expand All @@ -196,6 +205,7 @@ public static IMetaService getMetaService() {
}

public static IQuotaService getQuotaService() {
checkNotClosed();
if (quotaService == null) {
synchronized (ServiceContainer.class) {
if (quotaService == null) {
Expand All @@ -208,6 +218,7 @@ public static IQuotaService getQuotaService() {
}

public static CatalogMetadataService getCatalogMetadataService() {
checkNotClosed();
if (catalogMetadataService == null) {
synchronized (ServiceContainer.class) {
if (catalogMetadataService == null) {
Expand All @@ -220,6 +231,7 @@ public static CatalogMetadataService getCatalogMetadataService() {
}

public static FileInfoCacheService getFileInfoCacheService() {
checkNotClosed();
if (fileInfoCacheService == null) {
synchronized (ServiceContainer.class) {
if (fileInfoCacheService == null) {
Expand All @@ -232,6 +244,7 @@ public static FileInfoCacheService getFileInfoCacheService() {
}

public static ArcticTransactionService getArcticTransactionService() {
checkNotClosed();
if (arcticTransactionService == null) {
synchronized (ServiceContainer.class) {
if (arcticTransactionService == null) {
Expand All @@ -244,6 +257,7 @@ public static ArcticTransactionService getArcticTransactionService() {
}

public static ITableTaskHistoryService getTableTaskHistoryService() {
checkNotClosed();
if (tableTaskHistoryService == null) {
synchronized (ServiceContainer.class) {
if (tableTaskHistoryService == null) {
Expand All @@ -256,6 +270,7 @@ public static ITableTaskHistoryService getTableTaskHistoryService() {
}

public static ITableInfoService getTableInfoService() {
checkNotClosed();
if (tableInfoService == null) {
synchronized (ServiceContainer.class) {
if (tableInfoService == null) {
Expand All @@ -267,6 +282,7 @@ public static ITableInfoService getTableInfoService() {
}

public static ArcticTableMetastoreHandler getTableMetastoreHandler() {
checkNotClosed();
if (tableMetastoreHandler == null) {
synchronized (ServiceContainer.class) {
if (tableMetastoreHandler == null) {
Expand All @@ -278,6 +294,7 @@ public static ArcticTableMetastoreHandler getTableMetastoreHandler() {
}

public static RuntimeDataExpireService getRuntimeDataExpireService() {
checkNotClosed();
if (runtimeDataExpireService == null) {
synchronized (ServiceContainer.class) {
if (runtimeDataExpireService == null) {
Expand All @@ -289,6 +306,7 @@ public static RuntimeDataExpireService getRuntimeDataExpireService() {
}

public static AdaptHiveService getAdaptHiveService() {
checkNotClosed();
if (adaptHiveService == null) {
synchronized (AdaptHiveService.class) {
if (adaptHiveService == null) {
Expand All @@ -300,6 +318,7 @@ public static AdaptHiveService getAdaptHiveService() {
}

public static ISupportHiveSyncService getSupportHiveSyncService() {
checkNotClosed();
if (supportHiveSyncService == null) {
synchronized (ServiceContainer.class) {
if (supportHiveSyncService == null) {
Expand All @@ -312,6 +331,7 @@ public static ISupportHiveSyncService getSupportHiveSyncService() {
}

public static DDLTracerService getDdlTracerService() {
checkNotClosed();
if (ddlTracerService == null) {
synchronized (ServiceContainer.class) {
if (ddlTracerService == null) {
Expand All @@ -324,6 +344,7 @@ public static DDLTracerService getDdlTracerService() {
}

public static TerminalManager getTerminalManager() {
checkNotClosed();
if (terminalManager == null) {
synchronized (ServiceContainer.class) {
if (terminalManager == null) {
Expand All @@ -334,27 +355,8 @@ public static TerminalManager getTerminalManager() {
return terminalManager;
}

@VisibleForTesting
public static void setMetaService(IMetaService imetaService) {
metaService = imetaService;
}

@VisibleForTesting
public static void setFileInfoCacheService(FileInfoCacheService testFileInfoCacheService) {
fileInfoCacheService = testFileInfoCacheService;
}

@VisibleForTesting
public static void setOptimizeService(IOptimizeService optimizeService) {
ServiceContainer.optimizeService = optimizeService;
}

@VisibleForTesting
public static void setTableTaskHistoryService(ITableTaskHistoryService tableHistoryService) {
tableTaskHistoryService = tableHistoryService;
}

public static OptimizeExecuteService getOptimizeExecuteService() {
checkNotClosed();
if (optimizeExecuteService == null) {
synchronized (ServiceContainer.class) {
if (optimizeExecuteService == null) {
Expand All @@ -366,6 +368,7 @@ public static OptimizeExecuteService getOptimizeExecuteService() {
}

public static ContainerMetaService getContainerMetaService() {
checkNotClosed();
if (containerMetaService == null) {
synchronized (ServiceContainer.class) {
if (containerMetaService == null) {
Expand All @@ -377,6 +380,7 @@ public static ContainerMetaService getContainerMetaService() {
}

public static PlatformFileInfoService getPlatformFileInfoService() {
checkNotClosed();
if (platformFileInfoService == null) {
synchronized (ServiceContainer.class) {
if (platformFileInfoService == null) {
Expand All @@ -388,6 +392,7 @@ public static PlatformFileInfoService getPlatformFileInfoService() {
}

public static TableBlockerService getTableBlockerService() {
checkNotClosed();
if (tableBlockerService == null) {
synchronized (ServiceContainer.class) {
if (tableBlockerService == null) {
Expand All @@ -398,6 +403,12 @@ public static TableBlockerService getTableBlockerService() {
return tableBlockerService;
}

private static void checkNotClosed() {
if (closed) {
throw new IllegalStateException("ServiceContainer has been closed");
}
}

private static void close(Closeable closeable) {
if (closeable == null) {
return;
Expand All @@ -408,8 +419,15 @@ private static void close(Closeable closeable) {
LOG.warn("failed to close {}, ignore", closeable);
}
}

public static void reset() {
clear();
closed = false;
}

public static void clear() {
closed = true;

close(optimizeService);
optimizeService = null;

Expand Down

0 comments on commit bb38fb5

Please sign in to comment.