Status

Current state: Under Discussion

JIRA: 

Unable to render Jira issues macro, execution error. Unable to render Jira issues macro, execution error.

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

State restoration is a key procedure in Kafka Streams when processing tasks are migrated in a rebalance, as well for maintaining standby tasks for failure recoveries.

This proposal aims to expose more visibilities around this procedure to users, and is composed of two components: 1) augmenting the metrics related to restoration, 2) add new APIs for StateRestoreListener.

Public Interfaces

Below summarizes the public API changes in this KIP.

Restoration metrics

We propose add metrics both on the thread-level (default reporting level is INFO) as well as on the task level (default reporting level is DEBUG).

Note that we will have separate thread handling restoration procedures, and hence their thread id would be different from stream threads.

Thread-level metric tags are:

• type=stream-state-updater-metrics
• thread-id=[threadId]

Task-level metric tags are:

• type=stream-task-metrics
• thread-id=[threadId]
• task-id=[taskId]

The POC implementation of the proposed metrics can be found here: https://github.com/apache/kafka/pull/12391

active-restoring-tasks

standby-updating-tasks

active-paused-tasks

standby-paused-tasks

idle-ratio

restore-ratio

checkpoint-ratio

active-restore-records-rate

standby-update-records-rate

restore-call-rate

restore-total

restore-rate

restore-remaining-records-total

Along with these new metrics, we would also deprecate the metrics below:

standby-process-ratio

New Method in StateRestoreListener

When an active task starts restoration, StateStoreListener#onRestoreStart would be triggered. The restoring task could end in two possible ways:

1) Restoration completes and the task could now be processed normally with incoming stream records. At this time StateStoreListener#onRestoreEnd would be triggered. 

2) Restoration was paused before completes, e.g. since another rebalance is triggered and this task is suspended and potentially migrated out of the current host later. At this time no callbacks would be triggered.

We propose to cover the second case above with a new API, so that each onRestoreStart function would be paired with either an onRestoreEnd function or an onRestorePaused function. Note that if the suspended task was re-assigned back to the current host, another onRestoreStart would be triggered again.

public interface StateRestoreListener {

    void onRestoreStart(final TopicPartition topicPartition,
                        final String storeName,
                        final long startingOffset,
                        final long endingOffset);

    void onRestoreEnd(final TopicPartition topicPartition,
                      final String storeName,
                      final long totalRestored);

    ...

    /**
     * NEW FUNC. Method called when restoring the {@link StateStore} is suspended due to the task being suspended from the host.
     *           If the task was resumed after suspension and restoration continues, another {@link onRestoreStart} would be called. 
     */
    default void onRestoreSuspended(final TopicPartition topicPartition,
                                    final String storeName,
                                    final long totalRestored) {
        // do nothing
    } 
}

Compatibility, Deprecation, and Migration Plan

• The default implementation of the new onRestorePaused function would be a no-op, to maintain backward compatibilities.
• Deprecated metric would still be exposed, and only be removed in the next major release.

Rejected Alternatives

None.