Merge branch 'gerrit/neo' into 'gerrit/trinity'
Change-Id: I7ee726406bf54a998b3c33ed54844e2d6dc0c5c0
diff --git a/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/ActiveManager.java b/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/ActiveManager.java
index 636279c..2720510 100644
--- a/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/ActiveManager.java
+++ b/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/ActiveManager.java
@@ -104,15 +104,17 @@
}
public void handle(ActiveManagerMessage message) throws HyracksDataException {
- LOGGER.debug("NC handling {}({})({})", message.getKind(), message.getRuntimeId(), message.getDesc());
switch (message.getKind()) {
case STOP_ACTIVITY:
+ logHandle(Level.DEBUG, message);
stopRuntime(message);
break;
case REQUEST_STATS:
+ logHandle(Level.TRACE, message);
requestStats((ActiveStatsRequestMessage) message);
break;
case GENERIC_EVENT:
+ logHandle(Level.DEBUG, message);
deliverGenericEvent(message);
break;
default:
@@ -120,6 +122,10 @@
}
}
+ private void logHandle(Level level, ActiveManagerMessage message) {
+ LOGGER.log(level, "NC handling {}({})({})", message.getKind(), message.getRuntimeId(), message.getDesc());
+ }
+
private void deliverGenericEvent(ActiveManagerMessage message) throws HyracksDataException {
try {
ActiveRuntimeId runtimeId = message.getRuntimeId();
@@ -151,7 +157,6 @@
return;
}
String stats = runtime.getStats();
- LOGGER.debug("Sending stats response for {} ", runtimeId);
ActiveStatsResponse response = new ActiveStatsResponse(reqId, stats, null);
((NodeControllerService) serviceCtx.getControllerService()).sendRealTimeApplicationMessageToCC(
message.getCcId(), JavaSerializationUtils.serialize(response), null);
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
index 7487209..45af279 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
@@ -501,7 +501,7 @@
ICCMessageBroker messageBroker = (ICCMessageBroker) applicationCtx.getServiceContext().getMessageBroker();
AlgebricksAbsolutePartitionConstraint runtimeLocations = getLocations();
int partition = 0;
- LOGGER.log(Level.INFO, "Sending stop messages to {}", runtimeLocations);
+ LOGGER.log(Level.INFO, "sending stop messages to {}", runtimeLocations);
for (String location : runtimeLocations.getLocations()) {
ActiveRuntimeId runtimeId = getActiveRuntimeId(partition++);
messageBroker.sendApplicationMessageToNC(new ActiveManagerMessage(ActiveManagerMessage.Kind.STOP_ACTIVITY,
@@ -564,10 +564,10 @@
WaitForStateSubscriber subscriber;
Future<Void> suspendTask;
synchronized (this) {
- LOGGER.log(level, "{} Suspending entity {}", jobId, entityId);
- LOGGER.log(level, "{} Waiting for ongoing activities", jobId);
+ LOGGER.log(level, "{} suspending entity {}", jobId, entityId);
+ LOGGER.log(level, "{} waiting for ongoing activities", jobId);
waitForNonTransitionState();
- LOGGER.log(level, "{} Proceeding with suspension. Current state is {}", jobId, state);
+ LOGGER.log(level, "{} proceeding with suspension. current state is {}", jobId, state);
if (state == ActivityState.STOPPED) {
suspended = true;
return;
@@ -588,12 +588,12 @@
doSuspend(metadataProvider);
return null;
});
- LOGGER.log(level, "{} Suspension task has been submitted", jobId);
+ LOGGER.log(level, "{} suspension task has been submitted", jobId);
}
try {
- LOGGER.log(level, "{} Waiting for suspension task to complete", jobId);
+ LOGGER.log(level, "{} waiting for suspension task to complete", jobId);
suspendTask.get();
- LOGGER.log(level, "{} Waiting for state to become SUSPENDED or TEMPORARILY_FAILED", jobId);
+ LOGGER.log(level, "{} waiting for state to become SUSPENDED or TEMPORARILY_FAILED", jobId);
subscriber.sync();
suspended = true;
} catch (Exception e) {
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveNotificationHandler.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveNotificationHandler.java
index 4d654d5..5fe082b 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveNotificationHandler.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveNotificationHandler.java
@@ -95,7 +95,7 @@
Object property = jobSpecification.getProperty(ACTIVE_ENTITY_PROPERTY_NAME);
if (!(property instanceof EntityId)) {
if (property != null) {
- LOGGER.debug("{} is not an active job. job property={}", jobId, property);
+ LOGGER.debug("{} is not an ingestion job. job property={}", jobId, property);
}
return;
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/CancelQueryRequest.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/CancelQueryRequest.java
index 0b8a689..6570041 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/CancelQueryRequest.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/CancelQueryRequest.java
@@ -82,7 +82,7 @@
@Override
public String toString() {
- return String.format("%s(id=%s, uuid=%s, contextId=%s, node=%s)", getClass().getSimpleName(), reqId, uuid,
- contextId, nodeId);
+ return "CancelQueryRequest{from='" + nodeId + "', reqId=" + reqId + ", uuid='" + uuid + "', contextId='"
+ + contextId + "'}";
}
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/CancelQueryResponse.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/CancelQueryResponse.java
index a711b73..68d3430 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/CancelQueryResponse.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/CancelQueryResponse.java
@@ -51,6 +51,6 @@
@Override
public String toString() {
- return String.format("%s(id=%s, status=%s)", getClass().getSimpleName(), reqId, status);
+ return "CancelQueryResponse{reqId=" + reqId + ", status=" + status + '}';
}
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/job/JobManager.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/job/JobManager.java
index ac1a9d9..be3daae 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/job/JobManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/job/JobManager.java
@@ -393,7 +393,7 @@
run.setStartTime(System.currentTimeMillis());
run.setStartTimeZoneId(ZoneId.systemDefault().getId());
JobId jobId = run.getJobId();
- logJobCapacity(run, "running");
+ logJobCapacity(run, "running", Level.DEBUG);
activeRunMap.put(jobId, run);
run.setStatus(JobStatus.RUNNING, null);
executeJobInternal(run);
@@ -401,7 +401,7 @@
// Queue a job when the required capacity for the job is not met.
private void queueJob(JobRun jobRun) throws HyracksException {
- logJobCapacity(jobRun, "queueing");
+ logJobCapacity(jobRun, "queueing", Level.INFO);
jobRun.setStatus(JobStatus.PENDING, null);
jobQueue.add(jobRun);
}
@@ -437,10 +437,10 @@
private void releaseJobCapacity(JobRun jobRun) {
final JobSpecification job = jobRun.getJobSpecification();
jobCapacityController.release(job);
- logJobCapacity(jobRun, "released");
+ logJobCapacity(jobRun, "released", Level.DEBUG);
}
- private void logJobCapacity(JobRun jobRun, String jobStateDesc) {
+ private void logJobCapacity(JobRun jobRun, String jobStateDesc, Level lvl) {
IClusterCapacity requiredResources = jobRun.getJobSpecification().getRequiredClusterCapacity();
if (requiredResources == null) {
return;
@@ -451,7 +451,7 @@
return;
}
IReadOnlyClusterCapacity clusterCapacity = jobCapacityController.getClusterCapacity();
- LOGGER.info("{} {}, memory={}, cpu={}, (new) cluster memory={}, cpu={}, currently running={}, queued={}",
+ LOGGER.log(lvl, "{} {}, memory={}, cpu={}, (new) cluster memory={}, cpu={}, currently running={}, queued={}",
jobStateDesc, jobRun.getJobId(), requiredMemory, requiredCPUs,
clusterCapacity.getAggregatedMemoryByteSize(), clusterCapacity.getAggregatedCores(),
getRunningJobsCount(), jobQueue.size());
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/partitions/PartitionMatchMaker.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/partitions/PartitionMatchMaker.java
index 8f91944..6278693 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/partitions/PartitionMatchMaker.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/partitions/PartitionMatchMaker.java
@@ -150,7 +150,7 @@
}
public void removeUncommittedPartitions(Set<PartitionId> partitionIds, Set<TaskAttemptId> taIds, JobId jobId) {
- if (!partitionIds.isEmpty()) {
+ if (partitionIds != null && !partitionIds.isEmpty()) {
LOGGER.debug("Removing uncommitted partitions {}: {}", jobId, partitionIds);
}
IEntryFilter<PartitionDescriptor> filter =
@@ -167,7 +167,7 @@
}
public void removePartitionRequests(Set<PartitionId> partitionIds, Set<TaskAttemptId> taIds, JobId jobId) {
- if (!partitionIds.isEmpty()) {
+ if (partitionIds != null && !partitionIds.isEmpty()) {
LOGGER.debug("Removing partition requests {}: {}", jobId, partitionIds);
}
IEntryFilter<PartitionRequest> filter = o -> taIds.contains(o.getRequestingTaskAttemptId());
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/GetResultPartitionLocationsWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/GetResultPartitionLocationsWork.java
index d1d2269..1e34b96 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/GetResultPartitionLocationsWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/GetResultPartitionLocationsWork.java
@@ -28,6 +28,7 @@
import org.apache.hyracks.control.cc.result.IResultDirectoryService;
import org.apache.hyracks.control.common.work.IResultCallback;
import org.apache.hyracks.control.common.work.SynchronizableWork;
+import org.apache.logging.log4j.Level;
public class GetResultPartitionLocationsWork extends SynchronizableWork {
private final ClusterControllerService ccs;
@@ -68,4 +69,9 @@
public String toString() {
return getName() + ": JobId@" + jobId + " ResultSetId@" + rsId + " Known@" + Arrays.toString(knownRecords);
}
+
+ @Override
+ public Level logLevel() {
+ return Level.TRACE;
+ }
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/JobCleanupWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/JobCleanupWork.java
index f065940..6fe9909 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/JobCleanupWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/JobCleanupWork.java
@@ -28,6 +28,7 @@
import org.apache.hyracks.control.cc.job.JobRun;
import org.apache.hyracks.control.common.work.AbstractWork;
import org.apache.hyracks.control.common.work.IResultCallback;
+import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -52,6 +53,7 @@
@Override
public void run() {
+ LOGGER.info("cleaning up {} on NCs, status={}", jobId, status);
final JobRun jobRun = jobManager.get(jobId);
if (jobRun == null) {
LOGGER.debug("ignoring cleanup for unknown {}", jobId);
@@ -78,4 +80,9 @@
return getName() + ": JobId@" + jobId + " Status@" + status
+ (exceptions == null ? "" : " Exceptions@" + exceptions);
}
+
+ @Override
+ public Level logLevel() {
+ return Level.TRACE;
+ }
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/TaskFailureWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/TaskFailureWork.java
index 33d391f..48fd403 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/TaskFailureWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/TaskFailureWork.java
@@ -22,6 +22,7 @@
import org.apache.hyracks.api.dataflow.TaskAttemptId;
import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.api.util.ErrorMessageUtil;
import org.apache.hyracks.control.cc.ClusterControllerService;
import org.apache.hyracks.control.cc.job.IJobManager;
import org.apache.hyracks.control.cc.job.JobRun;
@@ -50,6 +51,7 @@
@Override
public String toString() {
- return getName() + ": [" + jobId + ":" + taId + ":" + nodeId + "]";
+ return getName() + ": [" + jobId + ":" + taId + ":" + nodeId + "] "
+ + ErrorMessageUtil.getCauseMessage(exceptions.get(0));
}
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/WaitForJobCompletionWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/WaitForJobCompletionWork.java
index ed3e574..63d5340 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/WaitForJobCompletionWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/work/WaitForJobCompletionWork.java
@@ -29,6 +29,7 @@
import org.apache.hyracks.control.cc.job.JobRun;
import org.apache.hyracks.control.common.work.IResultCallback;
import org.apache.hyracks.control.common.work.SynchronizableWork;
+import org.apache.logging.log4j.Level;
public class WaitForJobCompletionWork extends SynchronizableWork {
private final ClusterControllerService ccs;
@@ -92,4 +93,9 @@
public String toString() {
return getName() + " jobId:" + jobId;
}
+
+ @Override
+ public Level logLevel() {
+ return Level.TRACE;
+ }
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/CleanupJobletWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/CleanupJobletWork.java
index efc8467..2036d72 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/CleanupJobletWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/CleanupJobletWork.java
@@ -47,7 +47,7 @@
@Override
public void run() {
- LOGGER.debug("cleaning up {}", jobId);
+ LOGGER.debug("cleaning up {}, status:{}", jobId, status);
ncs.removeJobParameterByteStore(jobId);
ncs.getPartitionManager().jobCompleted(jobId, status);
Map<JobId, Joblet> jobletMap = ncs.getJobletMap();