rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1072022879
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6246,86 @@ public boolean isWrapperFor(Class<?> iface) throws
SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst)
throws MetaException, NoSuchCompactionException {
+ AbortCompactResponse response = new AbortCompactResponse(new
ArrayList<>());
+ List<Long> requestedCompId = reqst.getCompactionIds();
+ if (requestedCompId.isEmpty()) {
+ LOG.info("Compaction ids missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("ompaction ids missing in request.
No compactions to abort");
+ }
+ List<AbortCompactionResponseElement> abortCompactionResponseElementList =
new ArrayList<>();
+ for (int i = 0; i < requestedCompId.size(); i++) {
+ AbortCompactionResponseElement responseEle =
abortCompaction(requestedCompId.get(i));
+ abortCompactionResponseElementList.add(responseEle);
+ }
+ response.setAbortedcompacts(abortCompactionResponseElementList);
+ return response;
+ }
+
+ @RetrySemantics.SafeToRetry
+ public AbortCompactionResponseElement abortCompaction(Long compId) throws
MetaException {
+ try {
+ AbortCompactionResponseElement responseEle = new
AbortCompactionResponseElement();
+ responseEle.setCompactionIds(compId);
+ try (Connection dbConn =
getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex)) {
+ Optional<CompactionInfo> compactionInfo =
getCompactionByCompId(dbConn, compId);
+ if (compactionInfo.isPresent()) {
+ try (PreparedStatement pStmt =
dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ CompactionInfo ci = compactionInfo.get();
+ ci.errorMessage = "Compaction aborted by user";
+ ci.state = TxnStore.ABORTED_STATE;
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, ci,
getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}",
ci, updCount);
+ dbConn.rollback();
+ }
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS",
updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM
\"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, ci.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE <{}>");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}",
ci, updCount);
+ dbConn.rollback();
+ } else {
+ responseEle.setMessage("Successfully Aborted Compaction ");
+ responseEle.setStatus("Success");
+ dbConn.commit();
+ }
+ }
+ }
+ } else {
+ responseEle.setMessage("Compaction element not eligible for
cancellation");
+ responseEle.setStatus("Error");
+ }
+ } catch (SQLException e) {
+ LOG.error("Failed to abort compaction request");
+ checkRetryable(e, "abortCompaction(" + compId + ")");
+ responseEle.setMessage("Error while aborting compaction");
+ responseEle.setStatus("Error");
+ }
+ return responseEle;
+ } catch (RetryException e) {
+ return abortCompaction(compId);
+ }
+
+ }
+
+ private Optional<CompactionInfo> getCompactionByCompId(Connection dbConn,
Long compId) throws SQLException, MetaException {
Review Comment:
fixed.
--
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]
