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

!1421 [I4Y3TQ] Fix Polling Workers Sync Issue For Resume Flow 

Merge pull request !1421 from i-robot/pull324
This commit is contained in:
i-robot 2022-03-22 01:53:56 +00:00 committed by Gitee
parent be22256fb6 b3973fcb82
commit 698c4570df
No known key found for this signature in database
GPG Key ID: 173E9B9CA92EEF8F
9 changed files with 155 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
presto-hive/src/main/java/io/prestosql/plugin/hive/OrcFileWriterFactory.java
View File

@ -15,6 +15,7 @@ package io.prestosql.plugin.hive;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import io.airlift.log.Logger;
import io.prestosql.orc.OrcDataSink;
import io.prestosql.orc.OrcDataSource;
import io.prestosql.orc.OrcDataSourceId;
@ -61,6 +62,8 @@ import static java.util.stream.Collectors.toList;
public class OrcFileWriterFactory
implements HiveFileWriterFactory
{
private static final Logger log = Logger.get(OrcFileWriterFactory.class);
private final HdfsEnvironment hdfsEnvironment;
private final TypeManager typeManager;
private final NodeVersion nodeVersion;
@ -195,6 +198,7 @@ public class OrcFileWriterFactory
}
Callable<Void> rollbackAction = () -> {
log.debug("RollBack action to delete file %s", path);
fileSystem.delete(path, false);
return null;
};
@ -238,6 +242,7 @@ public class OrcFileWriterFactory
protected OrcDataSink createOrcDataSink(ConnectorSession session, FileSystem fileSystem, Path path)
throws IOException
{
log.debug("Creation of OrcDataSink for file %s", path);
return new OutputStreamOrcDataSink(fileSystem.create(path));
}

6
presto-main/src/main/java/io/prestosql/execution/SqlQueryExecution.java
View File

@ -607,7 +607,8 @@ public class SqlQueryExecution
heuristicIndexerManager,
snapshotManager,
// Require same number of tasks to be scheduled, but do not require it if starting from beginning
snapshotId.isPresent() ? queryScheduler.get().getStageTaskCounts() : null);
snapshotId.isPresent() ? queryScheduler.get().getStageTaskCounts() : null,
true);
}
private void resetOutputData(PlanRoot plan, OptionalLong snapshotId)
@ -868,7 +869,8 @@ public class SqlQueryExecution
dynamicFilterService,
heuristicIndexerManager,
snapshotManager,
null);
null,
false);
queryScheduler.set(scheduler);

5
presto-main/src/main/java/io/prestosql/execution/scheduler/NodeScheduler.java
View File

@ -363,4 +363,9 @@ public class NodeScheduler
.collect(toImmutableList());
return whenAnyCompleteCancelOthers(stateChangeFutures);
}
public void refreshNodeStates()
{
nodeManager.refreshWorkerStates();
}
}

21
presto-main/src/main/java/io/prestosql/execution/scheduler/SqlQueryScheduler.java
View File

@ -180,7 +180,8 @@ public class SqlQueryScheduler
DynamicFilterService dynamicFilterService,
HeuristicIndexerManager heuristicIndexerManager,
QuerySnapshotManager snapshotManager,
Map<StageId, Integer> stageTaskCounts)
Map<StageId, Integer> stageTaskCounts,
boolean isResume)
{
SqlQueryScheduler sqlQueryScheduler = new SqlQueryScheduler(
queryStateMachine,
@ -202,7 +203,8 @@ public class SqlQueryScheduler
dynamicFilterService,
heuristicIndexerManager,
snapshotManager,
stageTaskCounts);
stageTaskCounts,
isResume);
sqlQueryScheduler.initialize();
return sqlQueryScheduler;
}
@ -227,7 +229,8 @@ public class SqlQueryScheduler
DynamicFilterService dynamicFilterService,
HeuristicIndexerManager heuristicIndexerManager,
QuerySnapshotManager snapshotManager,
Map<StageId, Integer> stageTaskCounts)
Map<StageId, Integer> stageTaskCounts,
boolean isResumeScheduler)
{
this.queryStateMachine = requireNonNull(queryStateMachine, "queryStateMachine is null");
this.executionPolicy = requireNonNull(executionPolicy, "schedulerPolicyFactory is null");
@ -270,7 +273,8 @@ public class SqlQueryScheduler
stageLinkageBuilder,
isSnapshotEnabled,
snapshotManager,
stageTaskCounts);
stageTaskCounts,
isResumeScheduler);
SqlStageExecution rootStage = stageExecutions.get(0);
rootStage.setOutputBuffers(rootOutputBuffers);
@ -429,7 +433,8 @@ public class SqlQueryScheduler
ImmutableMap.Builder<StageId, StageLinkage> stageLinkages,
boolean isSnapshotEnabled,
QuerySnapshotManager snapshotManager,
Map<StageId, Integer> stageTaskCounts)
Map<StageId, Integer> stageTaskCounts,
boolean isResumeScheduler)
{
ImmutableList.Builder<SqlStageExecution> localStages = ImmutableList.builder();
@ -463,6 +468,9 @@ public class SqlQueryScheduler
if (isInternalSystemConnector(catalogName)) {
catalogName = null;
}
if (isResumeScheduler) {
nodeScheduler.refreshNodeStates();
}
NodeSelector nodeSelector = nodeScheduler.createNodeSelector(catalogName, keepConsumerOnFeederNodes, feederScheduledNodes);
if (isSnapshotEnabled) {
// When snapshot is enabled, then no task can be added after the query started running,
@ -596,7 +604,8 @@ public class SqlQueryScheduler
stageLinkages,
isSnapshotEnabled,
snapshotManager,
stageTaskCounts);
stageTaskCounts,
isResumeScheduler);
localStages.addAll(subTree);
SqlStageExecution childStage = subTree.get(0);

5
presto-main/src/main/java/io/prestosql/failuredetector/FailureDetector.java
View File

@ -24,6 +24,11 @@ public interface FailureDetector
State getState(HostAddress hostAddress);
default void waitForServiceStateRefresh()
{
// do - nothing
}
enum State
{
UNKNOWN,

107
presto-main/src/main/java/io/prestosql/failuredetector/HeartbeatFailureDetector.java
View File

@ -61,6 +61,7 @@ import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.collect.ImmutableList.toImmutableList;
@ -97,6 +98,7 @@ public class HeartbeatFailureDetector
private final boolean httpsRequired;
private final AtomicBoolean started = new AtomicBoolean();
private AtomicBoolean isTasksRemoved = new AtomicBoolean();
@Inject
public HeartbeatFailureDetector(
@ -221,32 +223,57 @@ public class HeartbeatFailureDetector
return builder.build();
}
@Override
public void waitForServiceStateRefresh()
{
Map<ServiceDescriptor, Long> waitingTasks = getTasksTimestamp();
long currentTime = waitingTasks.values().stream().mapToLong(t -> t).max().getAsLong();
updateMonitoredServices();
// remove expired tasks
synchronized (tasks) {
removeExpiredIds();
if (tasks.size() == 0) {
return;
}
}
Map<ServiceDescriptor, Long> finalCurrentTasks;
do {
Map<ServiceDescriptor, Long> currentTasks;
Set<ServiceDescriptor> failedTasks;
synchronized (tasks) {
currentTasks = getTasksTimestamp();
failedTasks = getFailed();
}
currentTasks.entrySet().stream().forEach(e -> {
if (e.getValue() > currentTime || failedTasks.contains(e.getKey())) {
if (currentTasks.containsKey(e.getKey())) {
currentTasks.remove(e.getKey());
}
}
});
finalCurrentTasks = currentTasks;
} while (finalCurrentTasks.size() > 0);
}
private ConcurrentMap<ServiceDescriptor, Long> getTasksTimestamp()
{
return tasks.values().stream().collect(Collectors.toConcurrentMap(t -> t.getService(), t -> t.getLastCompleteTimestamp()));
}
@VisibleForTesting
void updateMonitoredServices()
{
Set<ServiceDescriptor> online = selector.selectAllServices().stream()
.filter(descriptor -> !nodeInfo.getNodeId().equals(descriptor.getNodeId()))
.collect(toImmutableSet());
Set<ServiceDescriptor> online = getOnlineServiceDescriptors();
Set<UUID> onlineIds = online.stream()
.map(ServiceDescriptor::getId)
.collect(toImmutableSet());
Set<UUID> onlineIds = getOnlineIds(online);
// make sure only one thread is updating the registrations
synchronized (tasks) {
// 1. remove expired tasks
List<UUID> expiredIds = tasks.values().stream()
.filter(MonitoringTask::isExpired)
.map(MonitoringTask::getService)
.map(ServiceDescriptor::getId)
.collect(toImmutableList());
tasks.keySet().removeAll(expiredIds);
removeExpiredIds();
// 2. disable offline services
tasks.values().stream()
.filter(task -> !onlineIds.contains(task.getService().getId()))
.forEach(MonitoringTask::disable);
disableOfflineTasks(onlineIds);
// 3. create tasks for new services
Set<ServiceDescriptor> newServices = online.stream()
@ -268,6 +295,38 @@ public class HeartbeatFailureDetector
}
}
private Set<ServiceDescriptor> getOnlineServiceDescriptors()
{
return selector.selectAllServices().stream()
.filter(descriptor -> !nodeInfo.getNodeId().equals(descriptor.getNodeId()))
.collect(toImmutableSet());
}
private Set<UUID> getOnlineIds(Set<ServiceDescriptor> online)
{
return online.stream()
.map(ServiceDescriptor::getId)
.collect(toImmutableSet());
}
private void disableOfflineTasks(Set<UUID> onlineIds)
{
tasks.values().stream()
.filter(task -> !onlineIds.contains(task.getService().getId()))
.forEach(MonitoringTask::disable);
}
private void removeExpiredIds()
{
List<UUID> expiredIds = tasks.values().stream()
.filter(MonitoringTask::isExpired)
.map(MonitoringTask::getService)
.map(ServiceDescriptor::getId)
.collect(toImmutableList());
tasks.keySet().removeAll(expiredIds);
}
private URI getHttpUri(ServiceDescriptor descriptor)
{
String url = descriptor.getProperties().get(httpsRequired ? "https" : "http");
@ -298,6 +357,12 @@ public class HeartbeatFailureDetector
@GuardedBy("this")
private Long successTransitionTimestamp;
@GuardedBy("this")
private long lastCompleteTimestamp;
@GuardedBy("this")
private double lastFailureCount;
private MonitoringTask(ServiceDescriptor service, URI uri)
{
this.uri = uri;
@ -358,6 +423,11 @@ public class HeartbeatFailureDetector
Duration.nanosSince(successTransitionTimestamp).compareTo(warmupInterval) < 0; // are we within the warmup period?
}
public synchronized long getLastCompleteTimestamp()
{
return lastCompleteTimestamp;
}
private void ping()
{
try {
@ -396,6 +466,11 @@ public class HeartbeatFailureDetector
}
else if (successTransitionTimestamp == null) {
successTransitionTimestamp = System.nanoTime();
lastCompleteTimestamp = System.nanoTime();
lastFailureCount = stats.getRecentFailures();
}
else if (Duration.nanosSince(lastCompleteTimestamp).compareTo(new Duration(1, TimeUnit.SECONDS)) > 0 && stats.getRecentFailures() == lastFailureCount) {
lastCompleteTimestamp = System.nanoTime();
}
}
}

22
presto-main/src/main/java/io/prestosql/metadata/DiscoveryNodeManager.java
View File

@ -160,6 +160,28 @@ public final class DiscoveryNodeManager
pollWorkers();
}
public void refreshWorkerStates()
{
failureDetector.waitForServiceStateRefresh();
pollWorkers();
AllNodes allNodesOverDiscovery = getAllNodes();
Set<InternalNode> aliveNodes = ImmutableSet.<InternalNode>builder()
.addAll(allNodesOverDiscovery.getActiveNodes())
.addAll(allNodesOverDiscovery.getIsolatingNodes())
.addAll(allNodesOverDiscovery.getIsolatedNodes())
.addAll(allNodesOverDiscovery.getShuttingDownNodes())
.build();
ImmutableSet<String> aliveNodeIds = aliveNodes.stream()
.map(InternalNode::getNodeIdentifier)
.collect(toImmutableSet());
// Remove nodes that don't exist anymore
// Make a copy to materialize the set difference
Set<String> deadNodes = difference(nodeStates.keySet(), aliveNodeIds).immutableCopy();
nodeStates.keySet().removeAll(deadNodes);
}
private void pollWorkers()
{
AllNodes allNodesOverDiscovery = getAllNodes();

6
presto-main/src/main/java/io/prestosql/metadata/InMemoryNodeManager.java
View File

@ -143,4 +143,10 @@ public class InMemoryNodeManager
{
listeners.remove(requireNonNull(listener, "listener is null"));
}
@Override
public void refreshWorkerStates()
{
// no-op
}
}

2
presto-main/src/main/java/io/prestosql/metadata/InternalNodeManager.java
View File

@ -37,4 +37,6 @@ public interface InternalNodeManager
void addNodeChangeListener(Consumer<AllNodes> listener);
void removeNodeChangeListener(Consumer<AllNodes> listener);
void refreshWorkerStates();
}