Murtadha Hubail has uploaded this change for review. (
https://asterix-gerrit.ics.uci.edu/3432
Change subject: [NO ISSUE][OTH] Simplify ResultJobRecord APIs
......................................................................
[NO ISSUE][OTH] Simplify ResultJobRecord APIs
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
- Simplify ResultJobRecord APIs by allowing only a single ResultSetId
per ResultJobRecord (i.e. per job).
- Fail result partition registration when a job attempts to use
multiple ResultSetIds or inconsistent number of partitions and
log the inconsistency.
- Remove unused methods.
Change-Id: I37816efc92ee9f5e66f29ce74dec4c6c5bd07c6f
---
M
hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/result/ResultJobRecord.java
M
hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/result/ResultDirectoryService.java
2 files changed, 37 insertions(+), 39 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb
refs/changes/32/3432/1
diff --git
a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/result/ResultJobRecord.java
b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/result/ResultJobRecord.java
index b3b0706..eebfd9d 100644
---
a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/result/ResultJobRecord.java
+++
b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/result/ResultJobRecord.java
@@ -21,12 +21,12 @@
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import org.apache.hyracks.api.exceptions.ErrorCode;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
public class ResultJobRecord implements IResultStateRecord {
@@ -77,15 +77,13 @@
}
private static final long serialVersionUID = 1L;
-
+ private static final Logger LOGGER = LogManager.getLogger();
private final long timestamp;
private long jobStartTime;
private long jobEndTime;
private Status status;
-
+ private ResultSetId rsId;
private ResultSetMetaData resultSetMetaData;
-
- private Map<ResultSetId, ResultSetMetaData> resultSetMetadataMap = new
HashMap<>();
public ResultJobRecord() {
this.timestamp = System.nanoTime();
@@ -116,10 +114,6 @@
updateState(State.SUCCESS);
}
- public void fail(ResultSetId rsId, int partition) {
- getOrCreateDirectoryRecord(rsId, partition).fail();
- }
-
public void fail(List<Exception> exceptions) {
updateState(State.FAILED);
status.setExceptions(exceptions);
@@ -139,47 +133,40 @@
StringBuilder sb = new StringBuilder();
sb.append("{ \"status\": ").append(status.toString()).append(", ");
sb.append("\"timestamp\": ").append(timestamp).append(", ");
- sb.append("\"resultsets\":
").append(Arrays.toString(resultSetMetadataMap.entrySet().toArray())).append("
}");
+ sb.append("\"resultset\": ").append(resultSetMetaData).append(" }");
return sb.toString();
}
public synchronized void setResultSetMetaData(ResultSetId rsId,
IResultMetadata metadata, int nPartitions)
throws HyracksDataException {
- ResultSetMetaData rsMd = resultSetMetadataMap.get(rsId);
- if (rsMd == null) {
- final ResultSetMetaData resultSetMetaData = new
ResultSetMetaData(nPartitions, metadata);
- resultSetMetadataMap.put(rsId, resultSetMetaData);
- this.resultSetMetaData = resultSetMetaData;
- } else if (rsMd.getRecords().length != nPartitions) {
- throw
HyracksDataException.create(ErrorCode.INCONSISTENT_RESULT_METADATA,
rsId.toString());
+ if (this.rsId == null) {
+ this.rsId = rsId;
+ this.resultSetMetaData = new ResultSetMetaData(nPartitions,
metadata);
+ } else if (!this.rsId.equals(rsId) ||
resultSetMetaData.getRecords().length != nPartitions) {
+ logInconsistentMetadata(rsId, nPartitions);
+ throw
HyracksDataException.create(ErrorCode.INCONSISTENT_RESULT_METADATA,
this.rsId.toString());
}
- //TODO(tillw) throwing a HyracksDataException here hangs the execution
tests
}
- public ResultSetMetaData getResultSetMetaData(ResultSetId rsId) {
- return resultSetMetadataMap.get(rsId);
- }
-
- public synchronized ResultDirectoryRecord
getOrCreateDirectoryRecord(ResultSetId rsId, int partition) {
- ResultDirectoryRecord[] records =
getResultSetMetaData(rsId).getRecords();
+ public synchronized ResultDirectoryRecord getOrCreateDirectoryRecord(int
partition) {
+ ResultDirectoryRecord[] records = resultSetMetaData.getRecords();
if (records[partition] == null) {
records[partition] = new ResultDirectoryRecord();
}
return records[partition];
}
- public synchronized ResultDirectoryRecord getDirectoryRecord(ResultSetId
rsId, int partition)
- throws HyracksDataException {
- ResultDirectoryRecord[] records =
getResultSetMetaData(rsId).getRecords();
+ public synchronized ResultDirectoryRecord getDirectoryRecord(int
partition) throws HyracksDataException {
+ ResultDirectoryRecord[] records = resultSetMetaData.getRecords();
if (records[partition] == null) {
throw HyracksDataException.create(ErrorCode.RESULT_NO_RECORD,
partition, rsId);
}
return records[partition];
}
- public synchronized void updateState(ResultSetId rsId) {
+ public synchronized void updateState() {
int successCount = 0;
- ResultDirectoryRecord[] records =
getResultSetMetaData(rsId).getRecords();
+ ResultDirectoryRecord[] records = resultSetMetaData.getRecords();
for (ResultDirectoryRecord record : records) {
if ((record != null) && (record.getStatus() ==
ResultDirectoryRecord.Status.SUCCESS)) {
successCount++;
@@ -193,4 +180,18 @@
public synchronized ResultSetMetaData getResultSetMetaData() {
return resultSetMetaData;
}
+
+ private void logInconsistentMetadata(ResultSetId rsId, int nPartitions) {
+ if (LOGGER.isWarnEnabled()) {
+ LOGGER.warn("inconsistent result metadata for result set {}",
this.rsId);
+ if (this.rsId.equals(rsId)) {
+ LOGGER.warn("inconsistent result set id. Current {}, new {}",
this.rsId, rsId);
+ }
+ final int expectedPartitions =
resultSetMetaData.getRecords().length;
+ if (expectedPartitions != nPartitions) {
+ LOGGER.warn("inconsistent result set number of partitions.
Current {}, new {}", expectedPartitions,
+ nPartitions);
+ }
+ }
+ }
}
diff --git
a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/result/ResultDirectoryService.java
b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/result/ResultDirectoryService.java
index a2218e2..bfecc48 100644
---
a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/result/ResultDirectoryService.java
+++
b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/result/ResultDirectoryService.java
@@ -121,7 +121,7 @@
throws HyracksDataException {
ResultJobRecord djr = getNonNullResultJobRecord(jobId);
djr.setResultSetMetaData(rsId, metadata, nPartitions);
- ResultDirectoryRecord record = djr.getOrCreateDirectoryRecord(rsId,
partition);
+ ResultDirectoryRecord record =
djr.getOrCreateDirectoryRecord(partition);
record.setNetworkAddress(networkAddress);
record.setEmpty(emptyResult);
@@ -147,8 +147,8 @@
public synchronized void reportResultPartitionWriteCompletion(JobId jobId,
ResultSetId rsId, int partition)
throws HyracksDataException {
ResultJobRecord djr = getNonNullResultJobRecord(jobId);
- djr.getDirectoryRecord(rsId, partition).writeEOS();
- djr.updateState(rsId);
+ djr.getDirectoryRecord(partition).writeEOS();
+ djr.updateState();
notifyAll();
}
@@ -178,7 +178,7 @@
@Override
public synchronized IResultMetadata getResultMetadata(JobId jobId,
ResultSetId rsId) throws HyracksDataException {
- return
getNonNullResultJobRecord(jobId).getResultSetMetaData(rsId).getMetadata();
+ return
getNonNullResultJobRecord(jobId).getResultSetMetaData().getMetadata();
}
@Override
@@ -228,7 +228,6 @@
private ResultDirectoryRecord[] updatedRecords(JobId jobId, ResultSetId
rsId, ResultDirectoryRecord[] knownRecords)
throws HyracksDataException {
ResultJobRecord djr = getNonNullResultJobRecord(jobId);
-
if (djr.getStatus().getState() == State.FAILED) {
List<Exception> caughtExceptions = djr.getStatus().getExceptions();
if (caughtExceptions != null && !caughtExceptions.isEmpty()) {
@@ -241,13 +240,11 @@
throw
HyracksDataException.create(ErrorCode.RESULT_FAILURE_NO_EXCEPTION, rsId, jobId);
}
}
-
- final ResultSetMetaData resultSetMetaData =
djr.getResultSetMetaData(rsId);
+ final ResultSetMetaData resultSetMetaData = djr.getResultSetMetaData();
if (resultSetMetaData == null) {
return null;
}
ResultDirectoryRecord[] records = resultSetMetaData.getRecords();
-
return Arrays.equals(records, knownRecords) ? null : records;
}
@@ -255,7 +252,7 @@
for (JobId jId : getJobIds()) {
pw.print(jId.toString());
pw.print(" - ");
- pw.println(String.valueOf(getResultJobRecord(jId)));
+ pw.println(getResultJobRecord(jId));
}
pw.flush();
return pw;
--
To view, visit https://asterix-gerrit.ics.uci.edu/3432
To unsubscribe, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-MessageType: newchange
Gerrit-Change-Id: I37816efc92ee9f5e66f29ce74dec4c6c5bd07c6f
Gerrit-Change-Number: 3432
Gerrit-PatchSet: 1
Gerrit-Owner: Murtadha Hubail <mhub...@apache.org>
