jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

[I3XYZK] fix cache lockup in refresh cycle

This commit is contained in:
nitin.kashyap 2021-06-26 23:15:35 +05:30
parent c49e5f77a8
commit f233a2599e
8 changed files with 70 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
presto-hive/src/main/java/io/prestosql/plugin/hive/ForCachingHiveMetastoreTableRefresh.java Normal file
View File

@ -0,0 +1,32 @@
/*
* Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.plugin.hive;
import javax.inject.Qualifier;
import java.lang.annotation.Retention;
import java.lang.annotation.Target;
import static java.lang.annotation.ElementType.FIELD;
import static java.lang.annotation.ElementType.METHOD;
import static java.lang.annotation.ElementType.PARAMETER;
import static java.lang.annotation.RetentionPolicy.RUNTIME;
@Retention(RUNTIME)
@Target({FIELD, PARAMETER, METHOD})
@Qualifier
public @interface ForCachingHiveMetastoreTableRefresh
{
}

2
presto-hive/src/main/java/io/prestosql/plugin/hive/HiveConfig.java
View File

@ -502,7 +502,7 @@ public class HiveConfig
return this;
}
@Min(1)
@Min(10)
public int getMaxMetastoreRefreshThreads()
{
return maxMetastoreRefreshThreads;

12
presto-hive/src/main/java/io/prestosql/plugin/hive/HiveModule.java
View File

@ -187,7 +187,17 @@ public class HiveModule
{
return new BoundedExecutor(
newCachedThreadPool(daemonThreadsNamed("hive-metastore-" + catalogName + "-%s")),
hiveConfig.getMaxMetastoreRefreshThreads());
(int) Math.max(hiveConfig.getMaxMetastoreRefreshThreads() * 0.9, 9));
}
@ForCachingHiveMetastoreTableRefresh
@Singleton
@Provides
public Executor createCachingHiveMetastoreTableRefreshExecutor(HiveCatalogName catalogName, HiveConfig hiveConfig)
{
return new BoundedExecutor(
newCachedThreadPool(daemonThreadsNamed("hive-metastore-refresh-" + catalogName + "-%s")),
(int) Math.max(hiveConfig.getMaxMetastoreRefreshThreads() * 0.1, 1));
}
@Singleton

19
presto-hive/src/main/java/io/prestosql/plugin/hive/metastore/CachingHiveMetastore.java
View File

@ -24,6 +24,7 @@ import com.google.common.util.concurrent.UncheckedExecutionException;
import io.airlift.log.Logger;
import io.airlift.units.Duration;
import io.prestosql.plugin.hive.ForCachingHiveMetastore;
import io.prestosql.plugin.hive.ForCachingHiveMetastoreTableRefresh;
import io.prestosql.plugin.hive.HiveBasicStatistics;
import io.prestosql.plugin.hive.HiveConfig;
import io.prestosql.plugin.hive.HiveErrorCode;
@ -113,11 +114,16 @@ public class CachingHiveMetastore
private final boolean dontVerifyCacheEntry;
@Inject
public CachingHiveMetastore(@ForCachingHiveMetastore HiveMetastore delegate, @ForCachingHiveMetastore Executor executor, HiveConfig hiveConfig, NodeManager nodeManager)
public CachingHiveMetastore(@ForCachingHiveMetastore HiveMetastore delegate,
@ForCachingHiveMetastore Executor executor,
@ForCachingHiveMetastoreTableRefresh Executor tableRefreshExecutor,
HiveConfig hiveConfig,
NodeManager nodeManager)
{
this(
delegate,
executor,
tableRefreshExecutor,
hiveConfig.getMetastoreCacheTtl(),
hiveConfig.getMetastoreRefreshInterval(),
hiveConfig.getMetastoreDBCacheTtl(),
@ -126,14 +132,14 @@ public class CachingHiveMetastore
!(nodeManager.getCurrentNode().isCoordinator() || hiveConfig.getWorkerMetaStoreCacheEnabled()));
}
public CachingHiveMetastore(HiveMetastore delegate, Executor executor, Duration cacheTtl, Duration refreshInterval,
public CachingHiveMetastore(HiveMetastore delegate, Executor executor, Executor tableRefreshExecutor, Duration cacheTtl, Duration refreshInterval,
Duration tableCacheTtl, Duration tableRefreshInterval,
long maximumSize, boolean skipCache)
{
this(
delegate,
executor,
OptionalLong.of(cacheTtl.toMillis()),
tableRefreshExecutor, OptionalLong.of(cacheTtl.toMillis()),
refreshInterval.toMillis() >= cacheTtl.toMillis() ? OptionalLong.empty() : OptionalLong.of(refreshInterval.toMillis()),
OptionalLong.of(tableCacheTtl.toMillis()),
tableRefreshInterval.toMillis() >= tableCacheTtl.toMillis() ? OptionalLong.empty() : OptionalLong.of(tableRefreshInterval.toMillis()),
@ -147,6 +153,7 @@ public class CachingHiveMetastore
return new CachingHiveMetastore(
delegate,
newDirectExecutorService(),
newDirectExecutorService(),
OptionalLong.empty(),
OptionalLong.empty(),
OptionalLong.empty(),
@ -155,7 +162,7 @@ public class CachingHiveMetastore
false || delegate instanceof CachingHiveMetastore);
}
private CachingHiveMetastore(HiveMetastore delegate, Executor executor,
private CachingHiveMetastore(HiveMetastore delegate, Executor executor, Executor tableRefreshExecutor,
OptionalLong expiresAfterWriteMillisTable, OptionalLong refreshMillsTable,
OptionalLong expiresAfterWriteMillisDB, OptionalLong refreshMillsDB,
long maximumSize, boolean skipCache)
@ -197,10 +204,10 @@ public class CachingHiveMetastore
.build(asyncReloading(CacheLoader.from(this::loadAllViews), executor));
tableCache = newCacheBuilder(tableCacheTtl, tableRefreshTtl, maximumSize)
.build(asyncReloading(CacheLoader.from(this::loadTable), executor));
.build(asyncReloading(CacheLoader.from(this::loadTable), tableRefreshExecutor));
partitionNamesCache = newCacheBuilder(tableCacheTtl, tableRefreshTtl, maximumSize)
.build(asyncReloading(CacheLoader.from(this::loadPartitionNames), executor));
.build(asyncReloading(CacheLoader.from(this::loadPartitionNames), tableRefreshExecutor));
tableStatisticsCache = newCacheBuilder(expiresAfterWriteMillisTable, refreshMillsTable, maximumSize)
.build(asyncReloading(new CacheLoader<WithIdentity<HiveTableName>, WithValidation<Table, PartitionStatistics>>()

4
presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHive.java
View File

@ -584,6 +584,7 @@ public abstract class AbstractTestHive
protected ConnectorPageSourceProvider pageSourceProvider;
protected ConnectorPageSinkProvider pageSinkProvider;
protected ExecutorService executor;
protected ExecutorService executorRefresh;
private ScheduledExecutorService heartbeatService;
private ScheduledExecutorService vacuumExecutorService;
@ -593,6 +594,7 @@ public abstract class AbstractTestHive
public void setupClass()
{
executor = newCachedThreadPool(daemonThreadsNamed("hive-%s"));
executorRefresh = newCachedThreadPool(daemonThreadsNamed("hive-refresh-%s"));
heartbeatService = newScheduledThreadPool(1);
vacuumExecutorService = newScheduledThreadPool(1);
hiveMetastoreClientService = newScheduledThreadPool(1);
@ -714,7 +716,7 @@ public abstract class AbstractTestHive
HiveMetastore metastore = new CachingHiveMetastore(
new BridgingHiveMetastore(new ThriftHiveMetastore(metastoreLocator, new ThriftHiveMetastoreConfig())),
executor,
Duration.valueOf("1m"),
executorRefresh, Duration.valueOf("1m"),
Duration.valueOf("15s"),
Duration.valueOf("1m"),
Duration.valueOf("15s"),

7
presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHiveFileSystem.java
View File

@ -74,6 +74,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
@ -173,6 +174,7 @@ public abstract class AbstractTestHiveFileSystem
MetastoreLocator metastoreLocator = new TestingMetastoreLocator(config, host, port);
ExecutorService executor = newCachedThreadPool(daemonThreadsNamed("hive-%s"));
ExecutorService executorRefresh = newCachedThreadPool(daemonThreadsNamed("hive-refresh-%s"));
HivePartitionManager hivePartitionManager = new HivePartitionManager(TYPE_MANAGER, config);
HdfsConfiguration hdfsConfiguration = hdfsConfigurationProvider.apply(config);
@ -181,6 +183,7 @@ public abstract class AbstractTestHiveFileSystem
metastoreClient = new TestingHiveMetastore(
new BridgingHiveMetastore(new ThriftHiveMetastore(metastoreLocator, new ThriftHiveMetastoreConfig())),
executor,
executorRefresh,
config,
getBasePath(),
hdfsEnvironment);
@ -472,9 +475,9 @@ public abstract class AbstractTestHiveFileSystem
private final Path basePath;
private final HdfsEnvironment hdfsEnvironment;
public TestingHiveMetastore(HiveMetastore delegate, ExecutorService executor, HiveConfig hiveConfig, Path basePath, HdfsEnvironment hdfsEnvironment)
public TestingHiveMetastore(HiveMetastore delegate, ExecutorService executor, Executor executorRefresh, HiveConfig hiveConfig, Path basePath, HdfsEnvironment hdfsEnvironment)
{
super(delegate, executor, hiveConfig, new TestingNodeManager("fake-environment"));
super(delegate, executor, executorRefresh, hiveConfig, new TestingNodeManager("fake-environment"));
this.basePath = basePath;
this.hdfsEnvironment = hdfsEnvironment;
}

4
presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveWriterFactory.java
View File

@ -67,17 +67,19 @@ public class TestHiveWriterFactory
{
private ThriftMetastoreClient mockClient;
protected ExecutorService executor;
protected ExecutorService executorRefresh;
protected HiveMetastore metastore;
private void setUp()
{
mockClient = new MockThriftMetastoreClient();
executor = newCachedThreadPool(daemonThreadsNamed("hive-%s"));
executorRefresh = newCachedThreadPool(daemonThreadsNamed("hive-refresh-%s"));
MetastoreLocator metastoreLocator = new MockMetastoreLocator(mockClient);
metastore = new CachingHiveMetastore(
new BridgingHiveMetastore(new ThriftHiveMetastore(metastoreLocator, new ThriftHiveMetastoreConfig())),
executor,
Duration.valueOf("1m"),
executorRefresh, Duration.valueOf("1m"),
Duration.valueOf("15s"),
Duration.valueOf("1m"),
Duration.valueOf("15s"),

3
presto-hive/src/test/java/io/prestosql/plugin/hive/metastore/TestCachingHiveMetastore.java
View File

@ -56,11 +56,12 @@ public class TestCachingHiveMetastore
mockClient = new MockThriftMetastoreClient();
MetastoreLocator metastoreLocator = new MockMetastoreLocator(mockClient);
ListeningExecutorService executor = listeningDecorator(newCachedThreadPool(daemonThreadsNamed("test-%s")));
ListeningExecutorService executorRefresh = listeningDecorator(newCachedThreadPool(daemonThreadsNamed("test-%s")));
ThriftHiveMetastore thriftHiveMetastore = new ThriftHiveMetastore(metastoreLocator, new ThriftHiveMetastoreConfig());
metastore = new CachingHiveMetastore(
new BridgingHiveMetastore(thriftHiveMetastore),
executor,
new Duration(5, TimeUnit.MINUTES),
executorRefresh, new Duration(5, TimeUnit.MINUTES),
new Duration(1, TimeUnit.MINUTES),
new Duration(5, TimeUnit.MINUTES),
new Duration(1, TimeUnit.MINUTES),