klcopp commented on a change in pull request #2642:
URL: https://github.com/apache/hive/pull/2642#discussion_r709137310
##########
File path: ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestWorker.java
##########
@@ -39,19 +39,15 @@
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
import org.apache.hadoop.hive.metastore.txn.TxnStore;
import org.apache.hadoop.hive.ql.io.AcidUtils;
+import org.apache.thrift.TException;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.DataInput;
-import java.io.DataInputStream;
-import java.io.DataOutput;
-import java.io.DataOutputStream;
+import java.io.*;
Review comment:
Nit: we prefer to not bulk import
##########
File path:
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -1104,89 +1104,101 @@ public boolean checkFailedCompactions(CompactionInfo
ci) throws MetaException {
@RetrySemantics.CannotRetry
public void markFailed(CompactionInfo ci) throws MetaException {//todo: this
should not throw
if (LOG.isDebugEnabled()) {
- LOG.debug("Marking as failed: CompactionInfo: " + ci.toString());
+ LOG.debug("Marking as failed: CompactionInfo: " + ((ci != null) ?
ci.toString() : " is missing"));
}
try {
Connection dbConn = null;
- Statement stmt = null;
PreparedStatement pStmt = null;
ResultSet rs = null;
- // the error message related to the failure is wrapped inside
CompactionInfo
- // fetch this info, since ci will be reused in subsequent queries
- String errorMessage = ci.errorMessage;
+
try {
dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED);
- stmt = dbConn.createStatement();
- pStmt = dbConn.prepareStatement("SELECT \"CQ_ID\", \"CQ_DATABASE\",
\"CQ_TABLE\", \"CQ_PARTITION\", "
- + "\"CQ_STATE\", \"CQ_TYPE\", \"CQ_TBLPROPERTIES\",
\"CQ_WORKER_ID\", \"CQ_START\", \"CQ_RUN_AS\", "
- + "\"CQ_HIGHEST_WRITE_ID\", \"CQ_META_INFO\",
\"CQ_HADOOP_JOB_ID\", \"CQ_ERROR_MESSAGE\", "
- + "\"CQ_ENQUEUE_TIME\", \"CQ_WORKER_VERSION\",
\"CQ_INITIATOR_ID\", \"CQ_INITIATOR_VERSION\" "
- + "FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?");
- pStmt.setLong(1, ci.id);
- rs = pStmt.executeQuery();
- if (rs.next()) {
- ci = CompactionInfo.loadFullFromCompactionQueue(rs);
- String s = "DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?";
- pStmt = dbConn.prepareStatement(s);
- pStmt.setLong(1, ci.id);
- LOG.debug("Going to execute update <" + s + ">");
- int updCnt = pStmt.executeUpdate();
- }
- else {
- if(ci.id > 0) {
- //the record with valid CQ_ID has disappeared - this is a sign of
something wrong
- throw new IllegalStateException("No record with CQ_ID=" + ci.id +
" found in COMPACTION_QUEUE");
- }
- }
- if(ci.id == 0) {
- //The failure occurred before we even made an entry in
COMPACTION_QUEUE
- //generate ID so that we can make an entry in COMPLETED_COMPACTIONS
- ci.id = generateCompactionQueueId(stmt);
- //mostly this indicates that the Initiator is paying attention to
some table even though
- //compactions are not happening.
+ if(ci == null) {
+ LOG.debug("CompactionInfo was null, creating a new one");
+ ci = new CompactionInfo();
+ ci.id = generateCompactionQueueId(dbConn);
+ ci.dbname = "unspecified db";
+ ci.tableName= "unspecified table";
ci.state = DID_NOT_INITIATE;
- //this is not strictly accurate, but 'type' cannot be null.
- if(ci.type == null) {
- ci.type = CompactionType.MINOR;
- }
+ ci.type = CompactionType.MINOR;
ci.start = getDbTime(dbConn);
- LOG.debug("The failure occurred before we even made an entry in
COMPACTION_QUEUE. Generated ID so that we "
- + "can make an entry in COMPLETED_COMPACTIONS. New Id: " +
ci.id);
+ ci.errorMessage = "Compaction Queue Information was missing";
} else {
- ci.state = FAILED_STATE;
+ LOG.debug("CompactionInfo present");
+ String errorMessage = ci.errorMessage;
+ pStmt = dbConn.prepareStatement("SELECT \"CQ_ID\", \"CQ_DATABASE\",
\"CQ_TABLE\", \"CQ_PARTITION\", "
+ + "\"CQ_STATE\", \"CQ_TYPE\", \"CQ_TBLPROPERTIES\",
\"CQ_WORKER_ID\", \"CQ_START\", \"CQ_RUN_AS\", "
+ + "\"CQ_HIGHEST_WRITE_ID\", \"CQ_META_INFO\",
\"CQ_HADOOP_JOB_ID\", \"CQ_ERROR_MESSAGE\", "
+ + "\"CQ_ENQUEUE_TIME\", \"CQ_WORKER_VERSION\",
\"CQ_INITIATOR_ID\", \"CQ_INITIATOR_VERSION\" "
+ + "FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?");
+ pStmt.setLong(1, ci.id);
+ rs = pStmt.executeQuery();
+ if (rs.next()) {
+ ci = CompactionInfo.loadFullFromCompactionQueue(rs);
+ String s = "DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?";
+ pStmt = dbConn.prepareStatement(s);
+ pStmt.setLong(1, ci.id);
+ LOG.debug("Going to execute update <" + s + ">");
+ int updCnt = pStmt.executeUpdate();
+ } else {
+ if (ci.id > 0) {
+ //the record with valid CQ_ID has disappeared - this is a sign
of something wrong
+ throw new IllegalStateException("No record with CQ_ID=" + ci.id
+ " found in COMPACTION_QUEUE");
+ }
+ }
+
+ if (ci.id == 0) {
+ //The failure occurred before we even made an entry in
COMPACTION_QUEUE
+ //generate ID so that we can make an entry in COMPLETED_COMPACTIONS
+ ci.id = generateCompactionQueueId(dbConn);
+ //mostly this indicates that the Initiator is paying attention to
some table even though
+ //compactions are not happening.
+ ci.state = DID_NOT_INITIATE;
+ //this is not strictly accurate, but 'type' cannot be null.
+ if (ci.type == null) {
+ ci.type = CompactionType.MINOR;
+ }
+ ci.start = getDbTime(dbConn);
+ LOG.debug("The failure occurred before we even made an entry in
COMPACTION_QUEUE. Generated ID so that we "
+ + "can make an entry in COMPLETED_COMPACTIONS. New Id: " +
ci.id);
+ } else {
+ ci.state = FAILED_STATE;
+ }
+
+ if(errorMessage != null) {
+ ci.errorMessage = errorMessage;
+ }
+
+ close(rs, pStmt, null);
}
- close(rs, stmt, null);
- closeStmt(pStmt);
pStmt = dbConn.prepareStatement("INSERT INTO \"COMPLETED_COMPACTIONS\"
"
+ "(\"CC_ID\", \"CC_DATABASE\", \"CC_TABLE\", \"CC_PARTITION\",
\"CC_STATE\", \"CC_TYPE\", "
+ "\"CC_TBLPROPERTIES\", \"CC_WORKER_ID\", \"CC_START\",
\"CC_END\", \"CC_RUN_AS\", "
+ "\"CC_HIGHEST_WRITE_ID\", \"CC_META_INFO\",
\"CC_HADOOP_JOB_ID\", \"CC_ERROR_MESSAGE\", "
+ "\"CC_ENQUEUE_TIME\", \"CC_WORKER_VERSION\",
\"CC_INITIATOR_ID\", \"CC_INITIATOR_VERSION\") "
+ "VALUES(?,?,?,?,?, ?,?,?,?,?, ?,?,?,?,?,?,?,?,?)");
- if (errorMessage != null) {
- ci.errorMessage = errorMessage;
- }
+
CompactionInfo.insertIntoCompletedCompactions(pStmt, ci,
getDbTime(dbConn));
int updCount = pStmt.executeUpdate();
LOG.debug("Inserted " + updCount + " entries into
COMPLETED_COMPACTIONS");
LOG.debug("Going to commit");
closeStmt(pStmt);
dbConn.commit();
} catch (SQLException e) {
- LOG.warn("markFailed(" + ci.id + "):" + e.getMessage());
+ LOG.warn("markFailed(" + ((ci != null) ? ci.id : "missing id") + "):"
+ e.getMessage());
LOG.debug("Going to rollback");
rollbackDBConn(dbConn);
checkRetryable(dbConn, e, "markFailed(" + ci + ")");
LOG.error("markFailed(" + ci + ") failed: " + e.getMessage(), e);
Review comment:
These will call ci#toString, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
