You are viewing a plain text version of this content. The canonical link for it is here.
Posted to gitbox@hive.apache.org by GitBox <gi...@apache.org> on 2021/09/15 13:14:18 UTC
[GitHub] [hive] klcopp commented on a change in pull request #2642: Hive 25518 compaction txn handle NPE fix
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org