Skip to content

Commit

Permalink
Merge pull request #2281 from HubSpot/deploy_bug_fixes
Browse files Browse the repository at this point in the history 
Various bug fixes and stability improvements
  • Loading branch information
@ssalinas
ssalinas authored Apr 6, 2022
2 parents a836084 + d4b6b68 commit 9b0cc5c
Show file tree
Hide file tree
Showing 13 changed files with 242 additions and 214 deletions.
14 changes: 9 additions & 5 deletions SingularityBase/src/main/java/com/hubspot/singularity/SingularityTaskIdHistory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@ public class SingularityTaskIdHistory implements Comparable<SingularityTaskIdHis
private final Optional<ExtendedTaskState> lastTaskState;
private final Optional<String> runId;

@SuppressFBWarnings("NP_NULL_PARAM_DEREF")
public static SingularityTaskIdHistory fromTaskIdAndTaskAndUpdates(
SingularityTaskId taskId,
SingularityTask task,
Expand All @@ -26,10 +25,15 @@ public static SingularityTaskIdHistory fromTaskIdAndTaskAndUpdates(
ExtendedTaskState lastTaskState = null;
long updatedAt = taskId.getStartedAt();

if (updates != null && !updates.isEmpty()) {
SingularityTaskHistoryUpdate lastUpdate = Collections.max(updates);
lastTaskState = lastUpdate.getTaskState();
updatedAt = lastUpdate.getTimestamp();
if (updates != null) {
Optional<SingularityTaskHistoryUpdate> maybeLastUpdate = updates
.stream()
.filter(Objects::nonNull)
.max(SingularityTaskHistoryUpdate::compareTo);
if (maybeLastUpdate.isPresent()) {
lastTaskState = maybeLastUpdate.get().getTaskState();
updatedAt = maybeLastUpdate.get().getTimestamp();
}
}

return new SingularityTaskIdHistory(
Expand Down
9 changes: 9 additions & 0 deletions ...ularityService/src/main/java/com/hubspot/singularity/config/SingularityConfiguration.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -458,6 +458,7 @@ public class SingularityConfiguration extends Configuration {
private boolean skipPersistingTooLongTaskIds = false;

private boolean allowEmptyRequestInstances = false;
private boolean verifyTaskDataWrites = false;

public long getAskDriverToKillTasksAgainAfterMillis() {
return askDriverToKillTasksAgainAfterMillis;
Expand Down Expand Up @@ -2151,4 +2152,12 @@ public boolean allowEmptyRequestInstances() {
public void setAllowEmptyRequestInstances(boolean allowEmptyRequestInstances) {
this.allowEmptyRequestInstances = allowEmptyRequestInstances;
}

public boolean isVerifyTaskDataWrites() {
return verifyTaskDataWrites;
}

public void setVerifyTaskDataWrites(boolean verifyTaskDataWrites) {
this.verifyTaskDataWrites = verifyTaskDataWrites;
}
}
259 changes: 100 additions & 159 deletions SingularityService/src/main/java/com/hubspot/singularity/data/CuratorAsyncManager.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
Expand Down Expand Up @@ -57,22 +58,16 @@ private <T> List<T> getAsyncChildrenThrows(
final Transcoder<T> transcoder
)
throws Exception {
try {
List<String> children = getChildren(parent);
final List<String> paths = Lists.newArrayListWithCapacity(children.size());

for (String child : children) {
paths.add(ZKPaths.makePath(parent, child));
}

List<T> result = new ArrayList<>(
getAsyncThrows(parent, paths, transcoder, Optional.empty()).values()
);
List<String> children = getChildren(parent);
final List<String> paths = Lists.newArrayListWithCapacity(children.size());

return result;
} catch (Throwable t) {
throw t;
for (String child : children) {
paths.add(ZKPaths.makePath(parent, child));
}

return new ArrayList<>(
getAsyncThrows(parent, paths, transcoder, Optional.empty()).values()
);
}

private <T> Map<String, T> getAsyncThrows(
Expand Down Expand Up @@ -104,29 +99,22 @@ private <T> Map<String, T> getAsyncThrows(
final CountDownLatch latch = new CountDownLatch(paths.size());
final AtomicInteger bytes = new AtomicInteger();

final BackgroundCallback callback = new BackgroundCallback() {

@Override
public void processResult(CuratorFramework client, CuratorEvent event)
throws Exception {
try {
if (event.getData() == null || event.getData().length == 0) {
LOG.trace("Expected active node {} but it wasn't there", event.getPath());
return;
}

bytes.getAndAdd(event.getData().length);
final T object = transcoder.fromBytes(event.getData());
synchronizedObjects.put(event.getPath(), object);

if (cache.isPresent()) {
cache.get().set(event.getPath(), object);
}
} catch (Exception e) {
LOG.error("Exception processing curator result", e);
} finally {
latch.countDown();
final BackgroundCallback callback = (client, event) -> {
try {
if (event.getData() == null || event.getData().length == 0) {
LOG.trace("Expected active node {} but it wasn't there", event.getPath());
return;
}

bytes.getAndAdd(event.getData().length);
final T object = transcoder.fromBytes(event.getData());
synchronizedObjects.put(event.getPath(), object);

cache.ifPresent(tZkCache -> tZkCache.set(event.getPath(), object));
} catch (Exception e) {
LOG.error("Exception processing curator result", e);
} finally {
latch.countDown();
}
};

Expand Down Expand Up @@ -169,25 +157,21 @@ private <T extends SingularityId> List<T> getChildrenAsIdsForParentsThrows(

final CountDownLatch latch = new CountDownLatch(parents.size());

final BackgroundCallback callback = new BackgroundCallback() {

@Override
public void processResult(CuratorFramework client, CuratorEvent event)
throws Exception {
try {
if (event.getChildren() == null || event.getChildren().size() == 0) {
LOG.trace("Expected children for node {} - but found none", event.getPath());
return;
}
synchronizedObjects.addAll(
Lists.transform(
event.getChildren(),
Transcoders.getFromStringFunction(idTranscoder)
)
);
} finally {
latch.countDown();
final BackgroundCallback callback = (client, event) -> {
try {
if (event.getChildren() == null || event.getChildren().size() == 0) {
LOG.trace("Expected children for node {} - but found none", event.getPath());
return;
}
synchronizedObjects.addAll(
event
.getChildren()
.stream()
.map(Transcoders.getFromStringFunction(idTranscoder))
.collect(Collectors.toList())
);
} finally {
latch.countDown();
}
};

Expand Down Expand Up @@ -218,10 +202,10 @@ protected <T extends SingularityId> List<T> getChildrenAsIds(
final String rootPath,
final IdTranscoder<T> idTranscoder
) {
return Lists.transform(
getChildren(rootPath),
Transcoders.getFromStringFunction(idTranscoder)
);
return getChildren(rootPath)
.stream()
.map(Transcoders.getFromStringFunction(idTranscoder))
.collect(Collectors.toList());
}

private <T extends SingularityId> List<T> existsThrows(
Expand All @@ -238,22 +222,17 @@ private <T extends SingularityId> List<T> existsThrows(

final CountDownLatch latch = new CountDownLatch(paths.size());

final BackgroundCallback callback = new BackgroundCallback() {

@Override
public void processResult(CuratorFramework client, CuratorEvent event)
throws Exception {
try {
if (event.getStat() != null) {
objects.add(
Transcoders
.getFromStringFunction(idTranscoder)
.apply(ZKPaths.getNodeFromPath(event.getPath()))
);
}
} finally {
latch.countDown();
final BackgroundCallback callback = (client, event) -> {
try {
if (event.getStat() != null) {
objects.add(
Transcoders
.getFromStringFunction(idTranscoder)
.apply(ZKPaths.getNodeFromPath(event.getPath()))
);
}
} finally {
latch.countDown();
}
};

Expand Down Expand Up @@ -293,18 +272,13 @@ private <T extends SingularityId> List<T> notExistsThrows(

final CountDownLatch latch = new CountDownLatch(pathsMap.size());

final BackgroundCallback callback = new BackgroundCallback() {

@Override
public void processResult(CuratorFramework client, CuratorEvent event)
throws Exception {
try {
if (event.getStat() == null) {
objects.add(pathsMap.get(event.getPath()));
}
} finally {
latch.countDown();
final BackgroundCallback callback = (client, event) -> {
try {
if (event.getStat() == null) {
objects.add(pathsMap.get(event.getPath()));
}
} finally {
latch.countDown();
}
};

Expand Down Expand Up @@ -352,25 +326,7 @@ protected <T> List<T> getAsync(
) {
try {
return new ArrayList<>(
getAsyncThrows(pathNameForLogs, paths, transcoder, Optional.<ZkCache<T>>empty())
.values()
);
} catch (Throwable t) {
throw new RuntimeException(t);
}
}

protected <T> Map<String, T> getAsyncWithPath(
final String pathNameForLogs,
final Collection<String> paths,
final Transcoder<T> transcoder
) {
try {
return getAsyncThrows(
pathNameForLogs,
paths,
transcoder,
Optional.<ZkCache<T>>empty()
getAsyncThrows(pathNameForLogs, paths, transcoder, Optional.empty()).values()
);
} catch (Throwable t) {
throw new RuntimeException(t);
Expand Down Expand Up @@ -404,24 +360,19 @@ protected <T> List<T> getAsyncNestedChildrenAsListThrows(
final List<T> results = new ArrayList<>();
final CountDownLatch latch = new CountDownLatch(allPaths.size());
final AtomicInteger bytes = new AtomicInteger();
final BackgroundCallback callback = new BackgroundCallback() {

@Override
public void processResult(CuratorFramework client, CuratorEvent event)
throws Exception {
try {
if (event.getData() == null || event.getData().length == 0) {
LOG.trace("Expected active node {} but it wasn't there", event.getPath());
return;
}
bytes.getAndAdd(event.getData().length);

final T object = transcoder.fromBytes(event.getData());

results.add(object);
} finally {
latch.countDown();
final BackgroundCallback callback = (client, event) -> {
try {
if (event.getData() == null || event.getData().length == 0) {
LOG.trace("Expected active node {} but it wasn't there", event.getPath());
return;
}
bytes.getAndAdd(event.getData().length);

final T object = transcoder.fromBytes(event.getData());

results.add(object);
} finally {
latch.countDown();
}
};

Expand Down Expand Up @@ -468,27 +419,22 @@ protected <T, Q> Map<T, List<Q>> getAsyncNestedChildDataAsMapThrows(
final ConcurrentHashMap<T, List<Q>> resultsMap = new ConcurrentHashMap<>();
final CountDownLatch latch = new CountDownLatch(allPathsMap.size());
final AtomicInteger bytes = new AtomicInteger();
final BackgroundCallback callback = new BackgroundCallback() {

@Override
public void processResult(CuratorFramework client, CuratorEvent event)
throws Exception {
try {
if (event.getData() == null || event.getData().length == 0) {
LOG.trace("Expected active node {} but it wasn't there", event.getPath());
return;
}
bytes.getAndAdd(event.getData().length);

final Q object = transcoder.fromBytes(event.getData());

if (allPathsMap.get(event.getPath()) != null) {
resultsMap.putIfAbsent(allPathsMap.get(event.getPath()), new ArrayList<Q>());
resultsMap.get(allPathsMap.get(event.getPath())).add(object);
}
} finally {
latch.countDown();
final BackgroundCallback callback = (client, event) -> {
try {
if (event.getData() == null || event.getData().length == 0) {
LOG.trace("Expected active node {} but it wasn't there", event.getPath());
return;
}
bytes.getAndAdd(event.getData().length);

final Q object = transcoder.fromBytes(event.getData());

if (allPathsMap.get(event.getPath()) != null) {
resultsMap.putIfAbsent(allPathsMap.get(event.getPath()), new ArrayList<>());
resultsMap.get(allPathsMap.get(event.getPath())).add(object);
}
} finally {
latch.countDown();
}
};

Expand Down Expand Up @@ -535,24 +481,19 @@ protected <T extends SingularityId> List<T> getAsyncNestedChildIdsAsListThrows(
final List<T> results = new ArrayList<>();
final CountDownLatch latch = new CountDownLatch(allPaths.size());
final AtomicInteger bytes = new AtomicInteger();
final BackgroundCallback callback = new BackgroundCallback() {

@Override
public void processResult(CuratorFramework client, CuratorEvent event)
throws Exception {
try {
event
.getChildren()
.forEach(
child -> {
final T object = transcoder.fromString(child);
bytes.getAndAdd(child.getBytes().length);
results.add(object);
}
);
} finally {
latch.countDown();
}
final BackgroundCallback callback = (client, event) -> {
try {
event
.getChildren()
.forEach(
child -> {
final T object = transcoder.fromString(child);
bytes.getAndAdd(child.getBytes().length);
results.add(object);
}
);
} finally {
latch.countDown();
}
};

Expand Down Expand Up @@ -612,7 +553,7 @@ private <T> T queryAndReturnResultsThrows(
log(
method.operationType,
Optional.of(paths.size()),
bytes.get() > 0 ? Optional.of(bytes.get()) : Optional.<Integer>empty(),
bytes.get() > 0 ? Optional.of(bytes.get()) : Optional.empty(),
start,
pathNameForLogs
);
Expand Down
Loading

0 comments on commit 9b0cc5c

Please sign in to comment.