/*
|
* Copyright 1999-2019 Seata.io Group.
|
*
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
* you may not use this file except in compliance with the License.
|
* You may obtain a copy of the License at
|
*
|
* http://www.apache.org/licenses/LICENSE-2.0
|
*
|
* Unless required by applicable law or agreed to in writing, software
|
* distributed under the License is distributed on an "AS IS" BASIS,
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
* See the License for the specific language governing permissions and
|
* limitations under the License.
|
*/
|
package io.seata.server.transaction.saga;
|
|
import io.netty.channel.Channel;
|
import io.seata.common.util.CollectionUtils;
|
import io.seata.core.exception.GlobalTransactionException;
|
import io.seata.core.exception.TransactionException;
|
import io.seata.core.model.BranchStatus;
|
import io.seata.core.model.BranchType;
|
import io.seata.core.model.GlobalStatus;
|
import io.seata.core.protocol.transaction.BranchCommitRequest;
|
import io.seata.core.protocol.transaction.BranchCommitResponse;
|
import io.seata.core.protocol.transaction.BranchRollbackRequest;
|
import io.seata.core.protocol.transaction.BranchRollbackResponse;
|
import io.seata.core.rpc.RemotingServer;
|
import io.seata.core.rpc.netty.ChannelManager;
|
import io.seata.server.coordinator.AbstractCore;
|
import io.seata.server.session.BranchSession;
|
import io.seata.server.session.GlobalSession;
|
import io.seata.server.session.SessionHelper;
|
import io.seata.server.session.SessionHolder;
|
|
import java.io.IOException;
|
import java.util.Map;
|
import java.util.concurrent.TimeoutException;
|
|
/**
|
* The type saga core.
|
*
|
* @author ph3636
|
*/
|
public class SagaCore extends AbstractCore {
|
|
public SagaCore(RemotingServer remotingServer) {
|
super(remotingServer);
|
}
|
|
@Override
|
public BranchType getHandleBranchType() {
|
return BranchType.SAGA;
|
}
|
|
@Override
|
public void globalSessionStatusCheck(GlobalSession globalSession) throws GlobalTransactionException {
|
// SAGA type accept forward(retry) operation on timeout or commit fail, forward operation will register remaining branches
|
}
|
|
@Override
|
public BranchStatus branchCommitSend(BranchCommitRequest request, GlobalSession globalSession,
|
BranchSession branchSession) throws IOException, TimeoutException {
|
Map<String, Channel> channels = ChannelManager.getRmChannels();
|
if (CollectionUtils.isEmpty(channels)) {
|
LOGGER.error("Failed to commit SAGA global[" + globalSession.getXid() + ", RM channels is empty.");
|
return BranchStatus.PhaseTwo_CommitFailed_Retryable;
|
}
|
String sagaResourceId = getSagaResourceId(globalSession);
|
Channel sagaChannel = channels.get(sagaResourceId);
|
if (sagaChannel == null) {
|
LOGGER.error("Failed to commit SAGA global[" + globalSession.getXid()
|
+ ", cannot find channel by resourceId[" + sagaResourceId + "]");
|
return BranchStatus.PhaseTwo_CommitFailed_Retryable;
|
}
|
BranchCommitResponse response = (BranchCommitResponse) remotingServer.sendSyncRequest(sagaChannel, request);
|
return response.getBranchStatus();
|
}
|
|
@Override
|
public BranchStatus branchRollbackSend(BranchRollbackRequest request, GlobalSession globalSession,
|
BranchSession branchSession) throws IOException, TimeoutException {
|
Map<String, Channel> channels = ChannelManager.getRmChannels();
|
if (CollectionUtils.isEmpty(channels)) {
|
LOGGER.error("Failed to rollback SAGA global[" + globalSession.getXid() + ", RM channels is empty.");
|
return BranchStatus.PhaseTwo_RollbackFailed_Retryable;
|
}
|
String sagaResourceId = getSagaResourceId(globalSession);
|
Channel sagaChannel = channels.get(sagaResourceId);
|
if (sagaChannel == null) {
|
LOGGER.error("Failed to rollback SAGA global[" + globalSession.getXid()
|
+ ", cannot find channel by resourceId[" + sagaResourceId + "]");
|
return BranchStatus.PhaseTwo_RollbackFailed_Retryable;
|
}
|
BranchRollbackResponse response = (BranchRollbackResponse) remotingServer.sendSyncRequest(sagaChannel, request);
|
return response.getBranchStatus();
|
}
|
|
@Override
|
public boolean doGlobalCommit(GlobalSession globalSession, boolean retrying) throws TransactionException {
|
try {
|
BranchStatus branchStatus = branchCommit(globalSession, SessionHelper.newBranch(BranchType.SAGA,
|
globalSession.getXid(), -1, getSagaResourceId(globalSession), globalSession.getStatus().name()));
|
|
switch (branchStatus) {
|
case PhaseTwo_Committed:
|
SessionHelper.removeAllBranch(globalSession, !retrying);
|
LOGGER.info("Successfully committed SAGA global[" + globalSession.getXid() + "]");
|
break;
|
case PhaseTwo_Rollbacked:
|
LOGGER.info("Successfully rollbacked SAGA global[" + globalSession.getXid() + "]");
|
SessionHelper.removeAllBranch(globalSession, !retrying);
|
SessionHelper.endRollbacked(globalSession, retrying);
|
return false;
|
case PhaseTwo_RollbackFailed_Retryable:
|
LOGGER.error("By [{}], failed to rollback SAGA global [{}], will retry later.", branchStatus,
|
globalSession.getXid());
|
SessionHolder.getRetryCommittingSessionManager().removeGlobalSession(globalSession);
|
globalSession.queueToRetryRollback();
|
return false;
|
case PhaseOne_Failed:
|
LOGGER.error("By [{}], finish SAGA global [{}]", branchStatus, globalSession.getXid());
|
SessionHelper.removeAllBranch(globalSession, !retrying);
|
globalSession.changeGlobalStatus(GlobalStatus.Finished);
|
globalSession.end();
|
return false;
|
case PhaseTwo_CommitFailed_Unretryable:
|
if (globalSession.canBeCommittedAsync()) {
|
LOGGER.error("By [{}], failed to commit SAGA global [{}]", branchStatus,
|
globalSession.getXid());
|
break;
|
} else {
|
SessionHelper.endCommitFailed(globalSession,retrying);
|
LOGGER.error("Finally, failed to commit SAGA global[{}]", globalSession.getXid());
|
return false;
|
}
|
default:
|
if (!retrying) {
|
globalSession.queueToRetryCommit();
|
} else {
|
LOGGER.error("Failed to commit SAGA global[{}], will retry later.", globalSession.getXid());
|
}
|
return false;
|
}
|
} catch (Exception ex) {
|
LOGGER.error("Failed to commit global[" + globalSession.getXid() + "]", ex);
|
|
if (!retrying) {
|
globalSession.queueToRetryRollback();
|
}
|
throw new TransactionException(ex);
|
}
|
return true;
|
}
|
|
@Override
|
public boolean doGlobalRollback(GlobalSession globalSession, boolean retrying) throws TransactionException {
|
try {
|
BranchStatus branchStatus = branchRollback(globalSession, SessionHelper.newBranch(BranchType.SAGA,
|
globalSession.getXid(), -1, getSagaResourceId(globalSession), globalSession.getStatus().name()));
|
|
switch (branchStatus) {
|
case PhaseTwo_Rollbacked:
|
SessionHelper.removeAllBranch(globalSession, !retrying);
|
LOGGER.info("Successfully rollbacked SAGA global[{}]",globalSession.getXid());
|
break;
|
case PhaseTwo_RollbackFailed_Unretryable:
|
SessionHelper.endRollbackFailed(globalSession, retrying);
|
LOGGER.error("Failed to rollback SAGA global[{}]", globalSession.getXid());
|
return false;
|
case PhaseTwo_CommitFailed_Retryable:
|
SessionHolder.getRetryRollbackingSessionManager().removeGlobalSession(globalSession);
|
globalSession.queueToRetryCommit();
|
LOGGER.warn("Retry by custom recover strategy [Forward] on timeout, SAGA global[{}]", globalSession.getXid());
|
return false;
|
default:
|
LOGGER.error("Failed to rollback SAGA global[{}]", globalSession.getXid());
|
if (!retrying) {
|
globalSession.queueToRetryRollback();
|
}
|
return false;
|
}
|
} catch (Exception ex) {
|
LOGGER.error("Failed to rollback global[{}]", globalSession.getXid(), ex);
|
if (!retrying) {
|
globalSession.queueToRetryRollback();
|
}
|
throw new TransactionException(ex);
|
}
|
return true;
|
}
|
|
@Override
|
public void doGlobalReport(GlobalSession globalSession, String xid, GlobalStatus globalStatus) throws TransactionException {
|
if (GlobalStatus.Committed.equals(globalStatus)) {
|
SessionHelper.removeAllBranch(globalSession, false);
|
SessionHelper.endCommitted(globalSession, false);
|
LOGGER.info("Global[{}] committed", globalSession.getXid());
|
} else if (GlobalStatus.Rollbacked.equals(globalStatus)
|
|| GlobalStatus.Finished.equals(globalStatus)) {
|
SessionHelper.removeAllBranch(globalSession, false);
|
SessionHelper.endRollbacked(globalSession, false);
|
LOGGER.info("Global[{}] rollbacked", globalSession.getXid());
|
} else {
|
globalSession.changeGlobalStatus(globalStatus);
|
LOGGER.info("Global[{}] reporting is successfully done. status[{}]", globalSession.getXid(), globalSession.getStatus());
|
|
if (GlobalStatus.RollbackRetrying.equals(globalStatus)
|
|| GlobalStatus.TimeoutRollbackRetrying.equals(globalStatus)
|
|| GlobalStatus.UnKnown.equals(globalStatus)) {
|
globalSession.queueToRetryRollback();
|
LOGGER.info("Global[{}] will retry rollback", globalSession.getXid());
|
} else if (GlobalStatus.CommitRetrying.equals(globalStatus)) {
|
globalSession.queueToRetryCommit();
|
LOGGER.info("Global[{}] will retry commit", globalSession.getXid());
|
}
|
}
|
}
|
|
/**
|
* get saga ResourceId
|
*
|
* @param globalSession the globalSession
|
* @return sagaResourceId
|
*/
|
private String getSagaResourceId(GlobalSession globalSession) {
|
return globalSession.getApplicationId() + "#" + globalSession.getTransactionServiceGroup();
|
}
|
}
|
