diff --git a/modules/core/src/main/java/org/apache/ignite/internal/logger/IgniteThrottledLoggerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/logger/IgniteThrottledLoggerImpl.java index a58e41612a46..949aded0482c 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/logger/IgniteThrottledLoggerImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/logger/IgniteThrottledLoggerImpl.java @@ -284,17 +284,29 @@ private void logInternal( return; } - String message = messageSupplier.get(); - - var msgKey = new LogThrottleKey(throwable, throttleKey == null ? message : throttleKey); + String message = null; + LogThrottleKey msgKey = null; while (true) { + if (msgKey == null) { + if (throttleKey == null) { + message = messageSupplier.get(); + msgKey = new LogThrottleKey(throwable, message); + } else { + msgKey = new LogThrottleKey(throwable, throttleKey); + } + } + Long loggedTs = messagesMap.get(msgKey); long curTs = FastTimestamps.coarseCurrentTimeMillis(); if (loggedTs == null || curTs - loggedTs >= throttleIntervalMs) { if (replace(msgKey, loggedTs, curTs)) { + if (message == null) { + message = messageSupplier.get(); + } + if (throwable == null) { delegate.log(level, message); } else { diff --git a/modules/raft/src/main/java/org/apache/ignite/internal/raft/client/LeaderAvailabilityState.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/client/LeaderAvailabilityState.java index 5dd46b1416e0..69804d2bbfa6 100644 --- a/modules/raft/src/main/java/org/apache/ignite/internal/raft/client/LeaderAvailabilityState.java +++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/client/LeaderAvailabilityState.java @@ -126,13 +126,17 @@ boolean updateKnownLeaderAndTerm(@Nullable Peer leader, long term) { synchronized (mutex) { if (stopped) { - LOG.debug("Ignoring leader update after stop [leader={}, term={}]", leader, term); + if (LOG.isDebugEnabled()) { + LOG.debug("Ignoring leader update after stop [leader={}, term={}]", leader, term); + } return false; } // Ignore stale term notifications. if (term <= currentTerm) { - LOG.debug("Ignoring stale leader [newTerm={}, currentTerm={}]", term, currentTerm); + if (LOG.isDebugEnabled()) { + LOG.debug("Ignoring stale leader [newTerm={}, currentTerm={}]", term, currentTerm); + } return false; } @@ -147,8 +151,10 @@ boolean updateKnownLeaderAndTerm(@Nullable Peer leader, long term) { futureToComplete = waiters; } - LOG.debug("Leader updated [leader={}, term={}, previousTerm={}, stateChange={}->{}]", - leader, term, previousTerm, previousState, currentState); + if (LOG.isDebugEnabled()) { + LOG.debug("Leader updated [leader={}, term={}, previousTerm={}, stateChange={}->{}]", + leader, term, previousTerm, previousState, currentState); + } } // Complete outside the lock to avoid potential deadlocks with future callbacks. @@ -212,8 +218,10 @@ void resetLeaderState() { currentState = State.WAITING_FOR_LEADER; waiters = new CompletableFuture<>(); - LOG.debug("Leader state reset [previousTerm={}, stateChange={}->{}]", - previousTerm, previousState, currentState); + if (LOG.isDebugEnabled()) { + LOG.debug("Leader state reset [previousTerm={}, stateChange={}->{}]", + previousTerm, previousState, currentState); + } } } @@ -243,8 +251,10 @@ void onGroupUnavailable(long termWhenDetected) { currentState = State.WAITING_FOR_LEADER; waiters = new CompletableFuture<>(); - LOG.debug("Group unavailable [term={}, stateChange={}->{}]", - termWhenDetected, previousState, currentState); + if (LOG.isDebugEnabled()) { + LOG.debug("Group unavailable [term={}, stateChange={}->{}]", + termWhenDetected, previousState, currentState); + } } } } diff --git a/modules/raft/src/main/java/org/apache/ignite/internal/raft/client/PhysicalTopologyAwareRaftGroupService.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/client/PhysicalTopologyAwareRaftGroupService.java index 6efa0bb79295..e8d0f6b9df92 100644 --- a/modules/raft/src/main/java/org/apache/ignite/internal/raft/client/PhysicalTopologyAwareRaftGroupService.java +++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/client/PhysicalTopologyAwareRaftGroupService.java @@ -216,7 +216,9 @@ private void requestLeaderManually( } if (leaderWithTerm == null || leaderWithTerm.leader() == null) { - LOG.debug("No leader information available [grp={}].", groupId()); + if (LOG.isDebugEnabled()) { + LOG.debug("No leader information available [grp={}].", groupId()); + } return; } @@ -309,11 +311,17 @@ private void finishSubscriptions() { CompletableFutures.allOf(futures).get(SUBSCRIPTION_CLEANUP_TIMEOUT_MS, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); - LOG.debug("Interrupted while waiting for subscription cleanup [grp={}].", groupId); + if (LOG.isDebugEnabled()) { + LOG.debug("Interrupted while waiting for subscription cleanup [grp={}].", groupId); + } } catch (ExecutionException e) { - LOG.debug("Error during subscription cleanup [grp={}].", groupId, e.getCause()); + if (LOG.isDebugEnabled()) { + LOG.debug("Error during subscription cleanup [grp={}].", groupId, e.getCause()); + } } catch (TimeoutException e) { - LOG.debug("Timeout waiting for subscription cleanup [grp={}].", groupId); + if (LOG.isDebugEnabled()) { + LOG.debug("Timeout waiting for subscription cleanup [grp={}].", groupId); + } } } diff --git a/modules/raft/src/main/java/org/apache/ignite/internal/raft/client/RaftCommandExecutor.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/client/RaftCommandExecutor.java index cba5b9e629da..e8e36e42ee1d 100644 --- a/modules/raft/src/main/java/org/apache/ignite/internal/raft/client/RaftCommandExecutor.java +++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/client/RaftCommandExecutor.java @@ -708,7 +708,9 @@ private void sendWithRetryWaitingForLeader( // Infinite wait mode: start a new retry phase. // We may not have probed all peers yet (e.g., with many peers and short retry timeout), // so we can't assume there's no leader. Start fresh and keep trying. - LOG.debug("Retry phase timeout expired with infinite deadline, starting new retry phase [groupId={}]", groupId); + if (LOG.isDebugEnabled()) { + LOG.debug("Retry phase timeout expired with infinite deadline, starting new retry phase [groupId={}]", groupId); + } Peer initialPeer = resolveInitialPeer(targetStrategy, originalPeer); if (initialPeer == null) { fut.completeExceptionally(new ReplicationGroupUnavailableException(groupId)); @@ -1016,7 +1018,9 @@ public void onAllPeersExhausted() { return; } - LOG.debug("All peers exhausted, waiting for leader [groupId={}, term={}]", groupId, termWhenStarted); + if (LOG.isDebugEnabled()) { + LOG.debug("All peers exhausted, waiting for leader [groupId={}, term={}]", groupId, termWhenStarted); + } leaderAvailabilityState.onGroupUnavailable(termWhenStarted); waitForLeaderAndRetry(futureInvokeResult, requestFactory, targetSelectionStrategy, originalPeer, deadline); } diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/NodeImpl.java b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/NodeImpl.java index 7e5be3b033f4..e6425e056dd4 100644 --- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/NodeImpl.java +++ b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/NodeImpl.java @@ -807,8 +807,10 @@ private boolean allowLaunchElection() { } if (this.electionTimeoutCounter == 1) { - LOG.debug("Node {} does not initiate leader election and waits for the next election timeout.", - getNodeId()); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} does not initiate leader election and waits for the next election timeout.", + getNodeId()); + } return false; } } @@ -1146,7 +1148,9 @@ public boolean init(final NodeOptions opts) { } if (this.snapshotExecutor != null && this.options.getSnapshotIntervalSecs() > 0) { - LOG.debug("Node {} start snapshot timer, term={}.", getNodeId(), this.currTerm); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} start snapshot timer, term={}.", getNodeId(), this.currTerm); + } this.snapshotTimer.start(); } @@ -1223,7 +1227,9 @@ private boolean initBallotBox() { long lastCommittedIndex = getLastCommittedIndexOnInit(); ballotBoxOpts.setLastCommittedIndex(lastCommittedIndex); - LOG.debug("Node {} init ballot box's lastCommittedIndex={}.", getNodeId(), lastCommittedIndex); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} init ballot box's lastCommittedIndex={}.", getNodeId(), lastCommittedIndex); + } return this.ballotBox.init(ballotBoxOpts); } @@ -1447,7 +1453,9 @@ private void electSelf() { return; } if (this.state == State.STATE_FOLLOWER) { - LOG.debug("Node {} stop election timer, term={}.", getNodeId(), this.currTerm); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} stop election timer, term={}.", getNodeId(), this.currTerm); + } this.electionTimer.stop(); } resetLeaderId(PeerId.emptyPeer(), new Status(RaftError.ERAFTTIMEDOUT, @@ -1455,7 +1463,9 @@ private void electSelf() { this.state = State.STATE_CANDIDATE; this.currTerm++; this.votedId = this.serverId.copy(); - LOG.debug("Node {} start vote timer, term={} .", getNodeId(), this.currTerm); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} start vote timer, term={} .", getNodeId(), this.currTerm); + } this.voteTimer.start(); this.voteCtx.init(this.conf.getConf(), this.conf.isStable() ? null : this.conf.getOldConf()); electSelfTerm = this.currTerm; @@ -1568,7 +1578,9 @@ private void becomeLeader() { if (peer.equals(this.serverId)) { continue; } - LOG.debug("Node {} add a replicator, term={}, peer={}.", getNodeId(), this.currTerm, peer); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} add a replicator, term={}, peer={}.", getNodeId(), this.currTerm, peer); + } if (!this.replicatorGroup.addReplicator(peer)) { LOG.error("Fail to add a replicator [node={}, peer={}].", getNodeId(), peer); } @@ -1576,7 +1588,9 @@ private void becomeLeader() { // Start learner's replicators for (final PeerId peer : this.conf.listLearners()) { - LOG.debug("Node {} add a learner replicator, term={}, peer={}.", getNodeId(), this.currTerm, peer); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} add a learner replicator, term={}, peer={}.", getNodeId(), this.currTerm, peer); + } if (!this.replicatorGroup.addReplicator(peer, ReplicatorType.Learner)) { LOG.error("Fail to add a learner replicator [node={}, peer={}].", getNodeId(), peer); } @@ -1597,8 +1611,10 @@ private void becomeLeader() { // should be in writeLock private void stepDown(final long term, final boolean wakeupCandidate, final Status status) { - LOG.debug("Node {} stepDown, term={}, newTerm={}, wakeupCandidate={}.", getNodeId(), this.currTerm, term, - wakeupCandidate); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} stepDown, term={}, newTerm={}, wakeupCandidate={}.", getNodeId(), this.currTerm, term, + wakeupCandidate); + } if (!this.state.isActive()) { return; } @@ -1653,7 +1669,9 @@ else if (this.state.compareTo(State.STATE_TRANSFERRING) <= 0) { this.electionTimer.restart(); } else { - LOG.debug("Node {} is a learner, election timer is not started.", this.getNodeId()); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} is a learner, election timer is not started.", this.getNodeId()); + } } } @@ -1722,7 +1740,9 @@ private void executeApplyingTasks(final List tasks) { State nodeState = this.state; if (nodeState != State.STATE_LEADER) { final Status st = cannotApplyBecauseNotLeaderStatus(nodeState); - LOG.debug("Node {} can't apply, status={}.", getNodeId(), st); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} can't apply, status={}.", getNodeId(), st); + } final List dones = tasks.stream().map(ele -> ele.done) .filter(Objects::nonNull).collect(Collectors.toList()); Utils.runInThread(this.getOptions().getCommonExecutor(), () -> { @@ -1740,8 +1760,10 @@ private void executeApplyingTasks(final List tasks) { final LogEntryAndClosure task = tasks.get(i); if (task.expectedTerm != -1 && task.expectedTerm != this.currTerm) { - LOG.debug("Node {} can't apply task whose expectedTerm={} doesn't match currTerm={}.", getNodeId(), - task.expectedTerm, this.currTerm); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} can't apply task whose expectedTerm={} doesn't match currTerm={}.", getNodeId(), + task.expectedTerm, this.currTerm); + } if (task.done != null) { final Status st = new Status(RaftError.EPERM, "expected_term=%d doesn't match current_term=%d", task.expectedTerm, this.currTerm); @@ -2807,7 +2829,9 @@ private void onCaughtUp(final PeerId peer, final long term, final long version, if (st.getCode() == RaftError.ETIMEDOUT.getNumber() && Utils.monotonicMs() - this.replicatorGroup.getLastRpcSendTimestamp(peer) <= this.options .getElectionTimeoutMs()) { - LOG.debug("Node {} waits peer {} to catch up.", getNodeId(), peer); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} waits peer {} to catch up.", getNodeId(), peer); + } final OnCaughtUp caughtUp = new OnCaughtUp(this, term, peer, version); final long dueTime = Utils.nowMs() + this.options.getElectionTimeoutMs(); if (this.replicatorGroup.waitCaughtUp(peer, this.options.getCatchupMargin(), dueTime, caughtUp)) { @@ -2913,8 +2937,10 @@ private void handleStepDownTimeout() { this.readLock.lock(); try { if (this.state.compareTo(State.STATE_TRANSFERRING) > 0) { - LOG.debug("Node {} stop step-down timer, term={}, state={}.", getNodeId(), this.currTerm, - this.state); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} stop step-down timer, term={}, state={}.", getNodeId(), this.currTerm, + this.state); + } return; } final long monotonicNowMs = Utils.monotonicMs(); @@ -2937,7 +2963,9 @@ private void handleStepDownTimeout() { this.writeLock.lock(); try { if (this.state.compareTo(State.STATE_TRANSFERRING) > 0) { - LOG.debug("Node {} stop step-down timer, term={}, state={}.", getNodeId(), this.currTerm, this.state); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} stop step-down timer, term={}, state={}.", getNodeId(), this.currTerm, this.state); + } return; } final long monotonicNowMs = Utils.monotonicMs(); @@ -3319,8 +3347,10 @@ public void handlePreVoteResponse(final PeerId peerId, final long term, final Re "Raft node receives higher term pre_vote_response.")); return; } - LOG.debug("Node {} received PreVoteResponse from {}, term={}, granted={}.", getNodeId(), peerId, - response.term(), response.granted()); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} received PreVoteResponse from {}, term={}, granted={}.", getNodeId(), peerId, + response.term(), response.granted()); + } // check granted quorum? if (response.granted()) { this.prevVoteCtx.grant(peerId); @@ -3368,7 +3398,9 @@ public void run(final Status status) { private void preVote() { long preVoteTerm; try { - LOG.debug("Node {} term {} start preVote.", getNodeId(), this.currTerm); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} term {} start preVote.", getNodeId(), this.currTerm); + } if (this.snapshotExecutor != null && this.snapshotExecutor.isInstallingSnapshot()) { LOG.warn( "Node {} term {} doesn't do preVote when installing snapshot as the configuration may be out of date.", @@ -3454,7 +3486,9 @@ private void handleVoteTimeout() { preVote(); } else { - LOG.debug("Node {} term {} retry to vote self.", getNodeId(), this.currTerm); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} term {} retry to vote self.", getNodeId(), this.currTerm); + } // unlock in electSelf electSelf(); } diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/Replicator.java b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/Replicator.java index 4e80cd423fd3..9efd8caa22ab 100644 --- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/Replicator.java +++ b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/Replicator.java @@ -849,8 +849,10 @@ public void run(final Status status) { addInflight(RequestType.AppendEntries, this.nextIndex, 0, 0, seq, rpcFuture); } - LOG.debug("Node {} send HeartbeatRequest to {} term {} lastCommittedIndex {}", this.options.getNode() - .getNodeId(), this.options.getPeerId(), this.options.getTerm(), request.committedIndex()); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} send HeartbeatRequest to {} term {} lastCommittedIndex {}", this.options.getNode() + .getNodeId(), this.options.getPeerId(), this.options.getTerm(), request.committedIndex()); + } } finally { unlockId(); @@ -1055,7 +1057,9 @@ void block(final long startTimeMs, @SuppressWarnings("unused") final int errorCo } final long dueTime = startTimeMs + this.options.getDynamicHeartBeatTimeoutMs(); try { - LOG.debug("Blocking nodeId {} for {} ms", this.options.getNode().getNodeId(), this.options.getDynamicHeartBeatTimeoutMs()); + if (LOG.isDebugEnabled()) { + LOG.debug("Blocking nodeId {} for {} ms", this.options.getNode().getNodeId(), this.options.getDynamicHeartBeatTimeoutMs()); + } this.blockTimer = this.timerManager.schedule(() -> onBlockTimeout(this.id, this.options.getCommonExecutor()), dueTime - Utils.nowMs(), TimeUnit.MILLISECONDS); this.statInfo.runningState = RunningState.BLOCKING; @@ -1138,9 +1142,11 @@ private void notifyOnCaughtUp(final int code, final boolean beforeDestroy) { if (this.nextIndex - 1 + this.catchUpClosure.getMaxMargin() < this.options.getLogManager() .getLastLogIndex()) { - LOG.debug("Catch up for in progress [node={}, currentIndex={}, leaderLogLastIndex={}, catchUpMargin={})].", - getOpts().getNode().getNodeId(), nextIndex - 1, options.getLogManager().getLastLogIndex(), - catchUpClosure.getMaxMargin()); + if (LOG.isDebugEnabled()) { + LOG.debug("Catch up for in progress [node={}, currentIndex={}, leaderLogLastIndex={}, catchUpMargin={})].", + getOpts().getNode().getNodeId(), nextIndex - 1, options.getLogManager().getLastLogIndex(), + catchUpClosure.getMaxMargin()); + } return; } @@ -1334,9 +1340,11 @@ void onRpcReturned(final ThreadId id, final RequestType reqType, final Status st } if (stateVersion != r.version) { - LOG.debug( - "Replicator {} ignored old version response {}, current version is {}, request is {}\n, and response is {}\n, status is {}.", - r, stateVersion, r.version, request, response, status); + if (LOG.isDebugEnabled()) { + LOG.debug( + "Replicator {} ignored old version response {}, current version is {}, request is {}\n, and response is {}\n, status is {}.", + r, stateVersion, r.version, request, response, status); + } id.unlock(); return; } @@ -1560,7 +1568,9 @@ private boolean onAppendEntriesReturned(final ThreadId id, final Inflight inflig r.resetInflights(); // prev_log_index and prev_log_term doesn't match if (response.lastLogIndex() + 1 < r.nextIndex) { - LOG.debug("LastLogIndex at node={} is {}", r.options.getNode().getNodeId(), response.lastLogIndex()); + if (LOG.isDebugEnabled()) { + LOG.debug("LastLogIndex at node={} is {}", r.options.getNode().getNodeId(), response.lastLogIndex()); + } // The peer contains less logs than leader r.nextIndex = response.lastLogIndex() + 1; } @@ -1568,7 +1578,9 @@ private boolean onAppendEntriesReturned(final ThreadId id, final Inflight inflig // The peer contains logs from old term which should be truncated, // decrease _last_log_at_peer by one to test the right index to keep if (r.nextIndex > 1) { - LOG.debug("Log index dismatch [node={}, logIndex={}].", r.options.getNode().getNodeId(), r.nextIndex); + if (LOG.isDebugEnabled()) { + LOG.debug("Log index dismatch [node={}, logIndex={}].", r.options.getNode().getNodeId(), r.nextIndex); + } r.nextIndex--; } else { @@ -1633,7 +1645,9 @@ private boolean fillCommonFields(final AppendEntriesRequestBuilder rb, long prev if (prevLogTerm == 0 && prevLogIndex != 0) { if (!isHeartbeat) { Requires.requireTrue(prevLogIndex < this.options.getLogManager().getFirstLogIndex()); - LOG.debug("Log was compacted [node={}, logIndex={}].", this.options.getNode().getNodeId(), prevLogIndex); + if (LOG.isDebugEnabled()) { + LOG.debug("Log was compacted [node={}, logIndex={}].", this.options.getNode().getNodeId(), prevLogIndex); + } rb.prevLogIndex(prevLogIndex); return false; } @@ -1655,7 +1669,9 @@ private boolean fillCommonFields(final AppendEntriesRequestBuilder rb, long prev private void waitMoreEntries(final long nextWaitIndex) { try { - LOG.debug("Node {} waits more entries", this.options.getNode().getNodeId()); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} waits more entries", this.options.getNode().getNodeId()); + } if (this.waitId >= 0) { return; } diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/PlacementDriverMessageProcessor.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/PlacementDriverMessageProcessor.java index f10be6a7bea4..aaf59cf1b215 100644 --- a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/PlacementDriverMessageProcessor.java +++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/PlacementDriverMessageProcessor.java @@ -289,7 +289,9 @@ private CompletableFuture waitForActualState(HybridTimestamp startTime, lo LOG.warn("Failed to read index from raft leader {}.", readIndexError, readIndexTimeTracker.timeMessageDetails()); } else { - LOG.debug("Successfully read index from raft leader {}.", readIndexTimeTracker.timeMessageDetails()); + if (LOG.isDebugEnabled()) { + LOG.debug("Successfully read index from raft leader {}.", readIndexTimeTracker.timeMessageDetails()); + } } }) .thenCompose(raftIndex -> { @@ -310,8 +312,10 @@ private CompletableFuture waitForActualState(HybridTimestamp startTime, lo LOG.warn("Failed to wait for storage index to reach raft leader {}.", storageIndexTrackerError, storageIndexUpdateTimeTracker.timeMessageDetails()); } else { - LOG.debug("Successfully waited for storage index to reach raft leader {}.", - storageIndexUpdateTimeTracker.timeMessageDetails()); + if (LOG.isDebugEnabled()) { + LOG.debug("Successfully waited for storage index to reach raft leader {}.", + storageIndexUpdateTimeTracker.timeMessageDetails()); + } } }); }); diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java index 72e9814750f3..0c1563098568 100644 --- a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java +++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java @@ -415,7 +415,13 @@ private void handleReplicaRequest(ReplicaRequest request, InternalClusterNode se if (indicatesUnexpectedProblem(ex)) { throttledLog.warn(THROTTLE_REQUEST_KEY, "{} [request={}].", ex, THROTTLE_REQUEST_KEY, request); } else { - throttledLog.debug(THROTTLE_REQUEST_KEY, "{} [request={}].", ex, THROTTLE_REQUEST_KEY, request); + if (throttledLog.isDebugEnabled()) { + throttledLog.debug( + THROTTLE_REQUEST_KEY, + () -> format("{} [request={}].", THROTTLE_REQUEST_KEY, request), + ex + ); + } } msg = prepareReplicaErrorResponse(sendTimestamp, ex); @@ -434,7 +440,9 @@ private void handleReplicaRequest(ReplicaRequest request, InternalClusterNode se res.delayedAckProcessor != null ? res.delayedAckProcessor : (res0, ex0) -> { NetworkMessage msg0; - LOG.debug("Sending delayed response for replica request [request={}]", request); + if (LOG.isDebugEnabled()) { + LOG.debug("Sending delayed response for replica request [request={}]", request); + } if (ex0 == null) { msg0 = prepareDelayedReplicaResponse(sendTimestamp, res0); diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaStateManager.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaStateManager.java index 066e33319a5b..5fb2e2ae2384 100644 --- a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaStateManager.java +++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaStateManager.java @@ -177,7 +177,9 @@ CompletableFuture weakStartReplica( synchronized (context) { ReplicaState state = context.replicaState; - LOG.debug("Weak replica start [grp={}, state={}, future={}].", groupId, state, context.previousOperationFuture); + if (LOG.isDebugEnabled()) { + LOG.debug("Weak replica start [grp={}, state={}, future={}].", groupId, state, context.previousOperationFuture); + } if (state == ReplicaState.STOPPED || state == ReplicaState.STOPPING) { return startReplica(groupId, context, startOperation); @@ -194,7 +196,9 @@ CompletableFuture weakStartReplica( } else if (state == ReplicaState.PRIMARY_ONLY) { context.replicaState = ReplicaState.ASSIGNED; - LOG.debug("Weak replica start complete [state={}].", context.replicaState); + if (LOG.isDebugEnabled()) { + LOG.debug("Weak replica start complete [state={}].", context.replicaState); + } return trueCompletedFuture(); } else if (state == ReplicaState.RESTART_PLANNED) { @@ -224,7 +228,9 @@ private CompletableFuture startReplica( } } - LOG.debug("Weak replica start complete [state={}, partitionStarted={}].", context.replicaState, partitionStarted); + if (LOG.isDebugEnabled()) { + LOG.debug("Weak replica start complete [state={}, partitionStarted={}].", context.replicaState, partitionStarted); + } return partitionStarted; })) @@ -258,8 +264,10 @@ CompletableFuture weakStopReplica( synchronized (context) { ReplicaState state = context.replicaState; - LOG.debug("Weak replica stop [grpId={}, state={}, reason={}, reservedForPrimary={}, future={}].", groupId, state, - reason, context.reservedForPrimary, context.previousOperationFuture); + if (LOG.isDebugEnabled()) { + LOG.debug("Weak replica stop [grpId={}, state={}, reason={}, reservedForPrimary={}, future={}].", groupId, state, + reason, context.reservedForPrimary, context.previousOperationFuture); + } if (reason == WeakReplicaStopReason.EXCLUDED_FROM_ASSIGNMENTS) { if (state == ReplicaState.ASSIGNED) { @@ -296,7 +304,9 @@ CompletableFuture weakStopReplica( } // State #RESTART_PLANNED is also no-op because replica will be stopped within deferred operation. - LOG.debug("Weak replica stop (sync part) complete [grpId={}, state={}].", groupId, context.replicaState); + if (LOG.isDebugEnabled()) { + LOG.debug("Weak replica stop (sync part) complete [grpId={}, state={}].", groupId, context.replicaState); + } return nullCompletedFuture(); } @@ -330,7 +340,9 @@ private CompletableFuture stopReplica( context.replicaState = ReplicaState.STOPPED; } - LOG.debug("Weak replica stop complete [grpId={}, state={}].", groupId, context.replicaState); + if (LOG.isDebugEnabled()) { + LOG.debug("Weak replica stop complete [grpId={}, state={}].", groupId, context.replicaState); + } return true; })) @@ -349,7 +361,9 @@ private CompletableFuture planDeferredReplicaStop( Supplier> deferredStopOperation ) { synchronized (context) { - LOG.debug("Planning deferred replica stop [groupId={}, reservedForPrimary={}].", groupId, context.reservedForPrimary); + if (LOG.isDebugEnabled()) { + LOG.debug("Planning deferred replica stop [groupId={}, reservedForPrimary={}].", groupId, context.reservedForPrimary); + } // No parallel actions affected this, continue. if (context.reservedForPrimary) { @@ -377,8 +391,10 @@ void reserveReplica(ReplicationGroupId groupId, HybridTimestamp leaseStartTime) ReplicaStateContext context = getContext(groupId); synchronized (context) { - LOG.debug("Trying to reserve replica [groupId={}, leaseStartTime={}, replicaState={}, reservedForPrimary={}].", - groupId, leaseStartTime, context.replicaState, context.reservedForPrimary); + if (LOG.isDebugEnabled()) { + LOG.debug("Trying to reserve replica [groupId={}, leaseStartTime={}, replicaState={}, reservedForPrimary={}].", + groupId, leaseStartTime, context.replicaState, context.reservedForPrimary); + } ReplicaState state = context.replicaState; @@ -525,7 +541,9 @@ void reserve(ReplicationGroupId groupId, HybridTimestamp leaseStartTime) { this.leaseStartTime)); } - LOG.debug("Reserving replica [groupId={}, leaseStartTime={}].", groupId, leaseStartTime); + if (LOG.isDebugEnabled()) { + LOG.debug("Reserving replica [groupId={}, leaseStartTime={}].", groupId, leaseStartTime); + } this.leaseStartTime = leaseStartTime; this.reservedForPrimary = true; diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/UpdatableTableImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/UpdatableTableImpl.java index 2cea64755920..8aabc85d5f73 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/UpdatableTableImpl.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/UpdatableTableImpl.java @@ -384,7 +384,9 @@ private CompletableFuture> handleInsertResults( /** Transforms keys list to appropriate exception. */ private static RuntimeException conflictKeysException(List conflictKeys) { - LOG.debug("Unable to insert rows because of conflict [rows={}]", conflictKeys); + if (LOG.isDebugEnabled()) { + LOG.debug("Unable to insert rows because of conflict [rows={}]", conflictKeys); + } return new SqlException(CONSTRAINT_VIOLATION_ERR, "PK unique constraint is violated"); } diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/ExecutionDistributionProviderImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/ExecutionDistributionProviderImpl.java index 675a7a8f4504..7f479d386132 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/ExecutionDistributionProviderImpl.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/ExecutionDistributionProviderImpl.java @@ -126,7 +126,9 @@ private CompletableFuture primaryReplica( return f.handle((primaryReplica, e) -> { if (e != null) { - LOG.debug("Failed to retrieve primary replica for partition {}", e, replicationGroupId); + if (LOG.isDebugEnabled()) { + LOG.debug("Failed to retrieve primary replica for partition {}", e, replicationGroupId); + } throw withCause(IgniteInternalException::new, REPLICA_UNAVAILABLE_ERR, "Failed to get the primary replica" + " [replicationGroupId=" + replicationGroupId + ']', e); diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/TablePartitionProcessor.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/TablePartitionProcessor.java index 4f8a0e7985cd..d517d61cad29 100644 --- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/TablePartitionProcessor.java +++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/TablePartitionProcessor.java @@ -517,16 +517,20 @@ private CommandResult handlePrimaryReplicaChangeCommand( long commandTerm ) { long storageLastAppliedIndex = storage.lastAppliedIndex(); - LOG.debug("Handling PrimaryReplicaChangeCommand [tableId={}, partId={}, commandIndex={}, storageLastAppliedIndex={}, " - + "leaseStartTime={}, primaryNodeId={}, primaryNodeName={}]", - storage.tableId(), storage.partitionId(), commandIndex, storageLastAppliedIndex, - cmd.leaseStartTime(), cmd.primaryReplicaNodeId(), cmd.primaryReplicaNodeName()); + if (LOG.isDebugEnabled()) { + LOG.debug("Handling PrimaryReplicaChangeCommand [tableId={}, partId={}, commandIndex={}, storageLastAppliedIndex={}, " + + "leaseStartTime={}, primaryNodeId={}, primaryNodeName={}]", + storage.tableId(), storage.partitionId(), commandIndex, storageLastAppliedIndex, + cmd.leaseStartTime(), cmd.primaryReplicaNodeId(), cmd.primaryReplicaNodeName()); + } // Skips the write command because the storage has already executed it. if (commandIndex <= storageLastAppliedIndex) { - LOG.debug("Skipping PrimaryReplicaChangeCommand - already applied [tableId={}, partId={}, commandIndex={}, " - + "storageLastAppliedIndex={}]", - storage.tableId(), storage.partitionId(), commandIndex, storageLastAppliedIndex); + if (LOG.isDebugEnabled()) { + LOG.debug("Skipping PrimaryReplicaChangeCommand - already applied [tableId={}, partId={}, commandIndex={}, " + + "storageLastAppliedIndex={}]", + storage.tableId(), storage.partitionId(), commandIndex, storageLastAppliedIndex); + } return EMPTY_NOT_APPLIED_RESULT; } @@ -540,8 +544,10 @@ private CommandResult handlePrimaryReplicaChangeCommand( return null; }); - LOG.debug("Successfully applied PrimaryReplicaChangeCommand [tableId={}, partId={}, commandIndex={}, leaseStartTime={}]", - storage.tableId(), storage.partitionId(), commandIndex, cmd.leaseStartTime()); + if (LOG.isDebugEnabled()) { + LOG.debug("Successfully applied PrimaryReplicaChangeCommand [tableId={}, partId={}, commandIndex={}, leaseStartTime={}]", + storage.tableId(), storage.partitionId(), commandIndex, cmd.leaseStartTime()); + } return EMPTY_APPLIED_RESULT; } diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/CheckCatalogVersionOnActionRequest.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/CheckCatalogVersionOnActionRequest.java index df9461a0e2d9..a9fcec00aa1f 100644 --- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/CheckCatalogVersionOnActionRequest.java +++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/CheckCatalogVersionOnActionRequest.java @@ -121,7 +121,9 @@ public CheckCatalogVersionOnActionRequest(CatalogService catalogService) { Status st = NodeImpl.cannotApplyBecauseNotLeaderStatus(state); - LOG.debug("Node {} can't apply, status={}.", node.getNodeId(), st); + if (LOG.isDebugEnabled()) { + LOG.debug("Node {} can't apply, status={}.", node.getNodeId(), st); + } PeerId leaderId = node.getLeaderId(); diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PlacementDriverHelper.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PlacementDriverHelper.java index 724912584a90..2d49258ff658 100644 --- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PlacementDriverHelper.java +++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PlacementDriverHelper.java @@ -111,7 +111,9 @@ private CompletableFuture awaitPrimaryReplicaWithExceptionHandling( return placementDriver.awaitPrimaryReplica(partitionId, timestamp, timeout, timeUnit) .handle((primaryReplica, e) -> { if (e != null) { - LOG.debug("Failed to retrieve primary replica for partition {}", partitionId, e); + if (LOG.isDebugEnabled()) { + LOG.debug("Failed to retrieve primary replica for partition {}", partitionId, e); + } throw withCause(TransactionException::new, REPLICA_UNAVAILABLE_ERR, "Failed to get the primary replica" diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistry.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistry.java index 66a50f1c8e4a..1c1fa661ded0 100644 --- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistry.java +++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistry.java @@ -138,8 +138,10 @@ private void abortTransaction(InternalTransaction tx) { LOG.error("Transaction abortion has failed {}.", ex, formatTxInfo(tx.id(), volatileTxStateMetaStorage)); } else { - LOG.debug("Transaction has aborted due to timeout {}.", - formatTxInfo(tx.id(), volatileTxStateMetaStorage)); + if (LOG.isDebugEnabled()) { + LOG.debug("Transaction has aborted due to timeout {}.", + formatTxInfo(tx.id(), volatileTxStateMetaStorage)); + } } }); } diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java index 595aef5b33e6..4637dd51b13e 100644 --- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java +++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java @@ -660,8 +660,10 @@ public CompletableFuture finish( Map enlistedGroups, UUID txId ) { - LOG.debug("Finish [commit={}, {}, groups={}, commitPartId={}].", commitIntent, - formatTxInfo(txId, txStateVolatileStorage, false), enlistedGroups, commitPartition); + if (LOG.isDebugEnabled()) { + LOG.debug("Finish [commit={}, {}, groups={}, commitPartId={}].", commitIntent, + formatTxInfo(txId, txStateVolatileStorage, false), enlistedGroups, commitPartition); + } assert enlistedGroups != null; @@ -871,8 +873,10 @@ private CompletableFuture durableFinish( } if (ReplicatorRecoverableExceptions.isRecoverable(cause)) { - LOG.debug("Failed to finish Tx. The operation will be retried {}.", ex, - formatTxInfo(txId, txStateVolatileStorage)); + if (LOG.isDebugEnabled()) { + LOG.debug("Failed to finish Tx. The operation will be retried {}.", ex, + formatTxInfo(txId, txStateVolatileStorage)); + } return supplyAsync(() -> durableFinish( observableTimestampTracker, commitPartition, @@ -906,9 +910,11 @@ private CompletableFuture sendFinishRequest( HybridTimestamp commitTimestamp, CompletableFuture txFinishFuture ) { - LOG.debug("Finish [partition={}, node={}, enlistmentConsistencyToken={}, commit={}, {}, groups={}", - commitPartition, primaryConsistentId, enlistmentConsistencyToken, commit, - formatTxInfo(txId, txStateVolatileStorage, false), enlistedPartitions); + if (LOG.isDebugEnabled()) { + LOG.debug("Finish [partition={}, node={}, enlistmentConsistencyToken={}, commit={}, {}, groups={}", + commitPartition, primaryConsistentId, enlistmentConsistencyToken, commit, + formatTxInfo(txId, txStateVolatileStorage, false), enlistedPartitions); + } return txMessageSender.finish( primaryConsistentId, diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/WriteIntentSwitchProcessor.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/WriteIntentSwitchProcessor.java index d89d88fc4695..666fc0c9bc2e 100644 --- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/WriteIntentSwitchProcessor.java +++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/WriteIntentSwitchProcessor.java @@ -103,9 +103,11 @@ public CompletableFuture switchWriteIntentsWith Throwable cause = ExceptionUtils.unwrapCause(ex); if (ReplicatorRecoverableExceptions.isRecoverable(cause)) { - LOG.debug("Failed to switch write intents for txn. The operation will be retried [{}, exception={}].", - formatTxInfo(txId, volatileTxStateMetaStorage, false), - ex.getClass().getSimpleName() + ": " + ex.getMessage()); + if (LOG.isDebugEnabled()) { + LOG.debug("Failed to switch write intents for txn. The operation will be retried [{}, exception={}].", + formatTxInfo(txId, volatileTxStateMetaStorage, false), + ex.getClass().getSimpleName() + ": " + ex.getMessage()); + } return switchWriteIntentsWithRetry(commit, commitTimestamp, txId, partition); }