RATIS-2681. Add a listener for gRPC log appender lifecycle

[HTHou]

What changes were proposed in this pull request?

Add an optional factory in GrpcConfigKeys.Server that creates a GrpcLogAppenderListener for each source group member and destination peer.

This PR is scoped to AppendEntries. A nested AppendEntries interface exposes request registration, received replies, INCONSISTENCY handling, local send failures, timeouts, response-stream errors and completion. The parent listener observes client resets and run-loop exit. InstallSnapshot observation is deferred to a separate change.

Each callback has a single invocation site. Request registration, client-reset and reply-inconsistency notifications use the existing appender write lock. Reply notifications do not acquire that lock and may race with resets or other terminal callbacks; consumers correlate requests and deduplicate outcomes. AppendEntries.onReplyInconsistency is separate from onResetClient: it is invoked after onReply for the inconsistent response, immediately before clearing pending requests, without resetting the client. Run-loop exit is reported in a finally block, including exceptional exits. Stream terminal callbacks are still delivered after appender stop or leadership change without updating replication state.

Ratis exposes lifecycle information only. Applications own message filtering, correlation, outcome interpretation, protection-method attribution and persistence. Factory, sub-listener accessor and callback exceptions are isolated. Callbacks must return promptly and must not retain payloads.

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/RATIS-2681

How was this patch tested?

• ./mvnw -o -pl ratis-test -am test checkstyle:check '-Dtest=TestGrpcLogAppenderCallbacks,TestGrpcLogAppenderListener,TestConfUtils,TestLogAppenderWithGrpc#testLogAppenderAutoRestartOnException,TestWatchRequestWithGrpc#testWatchRequestAsyncChangeLeader' -Dsurefire.failIfNoSpecifiedTests=false
• ./mvnw -o -pl ratis-grpc -am -DskipTests -Djacoco.skip spotbugs:check apache-rat:check

All 24 targeted tests pass locally on JDK 25: 16 listener tests, 6 configuration tests, and the existing automatic-appender-restart and Watch leader-change regressions. Coverage includes real three-peer replication with an injected send failure and successful retries, stream creation failures, timeouts, late replies, INCONSISTENCY callback ordering and separation from client reset, reply processing while another thread holds the appender write lock, factory/accessor/callback exception isolation, normal/exceptional run-loop exit, and stream callbacks after appender stop or leadership change. Checkstyle, SpotBugs and license checks pass.

[slfan1989]

I have one concern about AppendLogResponseHandler.onError().

When the log appender is no longer running, the method returns before clearing and reporting the pending requests:

if (!isRunning()) {
  return;
}

If a shutdown or leader transition races with a gRPC stream error, some already-sent data-bearing requests may never receive a terminal FAILURE event.

This seems inconsistent with the requirement that every transfer attempt should generate exactly one terminal event. The snapshot error path already reports the failure before checking isRunning().

Could we report the affected sent requests as failures before returning, without updating the Raft replication state?

It would also be helpful to add a regression test covering a stream failure with multiple pending data-bearing requests.

[HTHou]

Addressed in the updated implementation. A stopped AppendLogResponseHandler now emits onAppendEntriesReset(error) before returning, without resetting the client or updating follower/leader state. The regression test covers both appender stop and leadership change with multiple pending requests, including a throwing listener.

The extension now exposes generic per-appender lifecycle callbacks. Consumers correlate request IDs and deduplicate terminal outcomes; audit-specific filtering and aggregation are outside Ratis. The snapshot error path also releases its waiter when the appender has stopped. All 16 targeted tests pass.

[szetszwo]

@HTHou , thanks for working on this!

• Let's implement appendEntries listener first to make this change smaller.
• Let's add sub-interfaces for AppendEntries (and InstallSnapshot later).
• Reorganize the methods so each method only be called once. I think it will be easier to understand the methods. Please take a look.

See https://issues.apache.org/jira/secure/attachment/13084350/1577_review.patch

[szetszwo]

It becomes unused. Let's remove it.

[HTHou]

Removed the unused three-argument constructor in 80b96f6.

[szetszwo]

Let's include the ids:

        LOG.warn("{}: Failed to create gRPC log appender listener for {}",
            server.getMemberId(), f.getPeer().getId(), t);

[HTHou]

Updated in 80b96f6: the warning now includes the source group member and destination peer IDs.

[HTHou]

Thanks @szetszwo! Updated in 80b96f6.

• Scoped this PR to AppendEntries and moved its callbacks into the AppendEntries sub-interface. InstallSnapshot callbacks/tests are removed for a follow-up.
• Each callback now has one invocation site. Removed the unused constructor and included both peer identities in factory-failure logs.
• Kept two small safeguards around the suggested structure: exceptions from appendEntries() are isolated, and onNotRunning runs in finally so exceptional appender exits are also observable.
• Named the parent callback onReset rather than onResetClient: INCONSISTENCY handling also discards pending requests without resetting the client. The same notification helper runs under the appender write lock at both reset sites, avoiding lost outstanding attempts or application-side tracking leaks.
• Added LOG_APPENDER_LISTENER_FACTORY_CLASS to fix TestConfUtils and retained the stopped/stepped-down stream-callback regression.

22 targeted tests, Checkstyle, SpotBugs and RAT pass locally.

[szetszwo]

@HTHou , thanks for the update! Please see the comments inlined.

[szetszwo]

How about adding onReplyInconsistency for INCONSISTENCY reply? Reset client and INCONSISTENCY should be treated differently.

      notifyAppendEntriesListener(GrpcLogAppenderListener.AppendEntries::onReplyInconsistency);

[HTHou]

Agreed. Updated in 7c6604c: added AppendEntries.onReplyInconsistency() at the pending-request clear in updateNextIndex, and renamed the parent callback to onResetClient. INCONSISTENCY handling no longer emits a client-reset notification. The new callback follows onReply for that response and runs under the existing index/queue write lock. The regression verifies the callback order and that the two events remain separate.

[szetszwo]

Question: Why acquiring writeLock here?

[HTHou]

The extra write lock was intended to serialize reply notifications with reset notifications, not to protect ConcurrentHashMap.remove(). I have removed it in 7c6604c and updated the listener contract: replies may race with reset/inconsistency and other terminal callbacks, so consumers handle correlation and deduplication. A regression now holds the appender write lock on one thread while a SUCCESS reply is processed and reported on another thread. All 24 targeted tests and Checkstyle/SpotBugs/RAT checks pass locally on JDK 25.