Skip to content

HDDS-11695. SCM follower should not log NotLeaderException during Pipeline Report processing. #7430

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
adoroszlai merged 2 commits into from
Nov 15, 2024
Merged

HDDS-11695. SCM follower should not log NotLeaderException during Pipeline Report processing. #7430

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8cbb843
HDDS-11695. SCM follower should not log NotLeaderException during Pip…
nandakumar131 Nov 13, 2024
4177dbf
Review comment addressed.
nandakumar131 Nov 14, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 29 additions & 18 deletions ...s/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineReportHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import org.apache.hadoop.hdds.protocol.proto
.StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
import org.apache.hadoop.hdds.scm.events.SCMEvents;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
import org.apache.hadoop.hdds.scm.ha.SCMContext;
import org.apache.hadoop.hdds.scm.safemode.SafeModeManager;
import org.apache.hadoop.hdds.scm.server
Expand Down Expand Up @@ -91,35 +92,45 @@ public void onMessage(PipelineReportFromDatanode pipelineReportFromDatanode,
for (PipelineReport report : pipelineReport.getPipelineReportList()) {
try {
processPipelineReport(report, dn, publisher);
} catch (NotLeaderException ex) {
// Avoid NotLeaderException logging which happens when processing
// pipeline report on followers.
} catch (PipelineNotFoundException e) {
LOGGER.error("Could not find pipeline {}", report.getPipelineID());
handlePipelineNotFoundException(report, dn, publisher);
} catch (IOException | TimeoutException e) {
LOGGER.error("Could not process pipeline report={} from dn={}.",
report, dn, e);
// Ignore NotLeaderException logging which happens when processing
// pipeline report on followers.
if (!isNotLeaderException(e)) {
LOGGER.error("Could not process pipeline report={} from dn={}.",
report, dn, e);
}
}
}
}

protected void processPipelineReport(PipelineReport report,
DatanodeDetails dn, EventPublisher publisher)
throws IOException, TimeoutException {
PipelineID pipelineID = PipelineID.getFromProtobuf(report.getPipelineID());
Pipeline pipeline;
try {
pipeline = pipelineManager.getPipeline(pipelineID);
} catch (PipelineNotFoundException e) {
if (scmContext.isLeader()) {
LOGGER.info("Reported pipeline {} is not found", pipelineID);
SCMCommand< ? > command = new ClosePipelineCommand(pipelineID);
private void handlePipelineNotFoundException(final PipelineReport report,
final DatanodeDetails dn, final EventPublisher publisher) {
final PipelineID pipelineID = PipelineID.getFromProtobuf(report.getPipelineID());
LOGGER.info("Pipeline {}, reported by datanode {} is not found.", pipelineID, dn);
if (scmContext.isLeader()) {
try {
final SCMCommand<?> command = new ClosePipelineCommand(pipelineID);
command.setTerm(scmContext.getTermOfLeader());
publisher.fireEvent(SCMEvents.DATANODE_COMMAND,
new CommandForDatanode<>(dn.getUuid(), command));
} catch (NotLeaderException ex) {
// Do nothing if the leader has changed.
}
return;
}
}

private static boolean isNotLeaderException(final Exception e) {
return e instanceof SCMException && ((SCMException) e).getResult().equals(
SCMException.ResultCodes.SCM_NOT_LEADER);
}

protected void processPipelineReport(PipelineReport report,
DatanodeDetails dn, EventPublisher publisher)
throws IOException, TimeoutException {
final PipelineID pipelineID = PipelineID.getFromProtobuf(report.getPipelineID());
final Pipeline pipeline = pipelineManager.getPipeline(pipelineID);

setReportedDatanode(pipeline, dn);
setPipelineLeaderId(report, pipeline, dn);
Expand Down