Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java (Java)

Code before:

    private final ChangelogReader changelogReader;
    private final ConsumerRebalanceListener rebalanceListener;
    private final Consumer<byte[], byte[]> mainConsumer;
    private final Consumer<byte[], byte[]> restoreConsumer;
    private final Admin adminClient;
    private final TopologyMetadata topologyMetadata;
    private final java.util.function.Consumer<Long> cacheResizer;

    private BiConsumer<Throwable, Boolean> streamsUncaughtExceptionHandler;
    private final Runnable shutdownErrorHook;

    // These must be Atomic references as they are shared and used to signal between the assignor and the stream thread
    private final AtomicInteger assignmentErrorCode;
    private final AtomicLong nextProbingRebalanceMs;
    // recoverable errors (don't require killing thread) that we need to invoke the exception
    // handler for, eg MissingSourceTopicException with named topologies
    private final Queue<StreamsException> nonFatalExceptionsToHandle;

    private final Optional<StreamsRebalanceData> streamsRebalanceData;
    private final StreamsMetadataState streamsMetadataState;

    // These are used to signal from outside the stream thread, but the variables themselves are internal to the thread
    private final AtomicLong cacheResizeSize = new AtomicLong(-1L);
    private final AtomicReference<org.apache.kafka.streams.CloseOptions.GroupMembershipOperation> leaveGroupRequested =
        new AtomicReference<>(org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.LEAVE_GROUP);
    private final AtomicLong lastShutdownWarningTimestamp = new AtomicLong(0L);
    private final boolean eosEnabled;
    private final boolean stateUpdaterEnabled;
    private final boolean processingThreadsEnabled;

    private volatile long fetchDeadlineClientInstanceId = -1;
    private volatile KafkaFutureImpl<Uuid> mainConsumerInstanceIdFuture = new KafkaFutureImpl<>();
    private volatile KafkaFutureImpl<Uuid> restoreConsumerInstanceIdFuture = new KafkaFutureImpl<>();
    private volatile KafkaFutureImpl<Uuid> producerInstanceIdFuture = new KafkaFutureImpl<>();

    private Timer topicsReadyTimer;

    public static StreamThread create(final TopologyMetadata topologyMetadata,
                                      final StreamsConfig config,
                                      final KafkaClientSupplier clientSupplier,
                                      final Admin adminClient,
                                      final UUID processId,
                                      final String clientId,
                                      final StreamsMetricsImpl streamsMetrics,
                                      final Time time,
                                      final StreamsMetadataState streamsMetadataState,
                                      final long cacheSizeBytes,
                                      final StateDirectory stateDirectory,
                                      final StateRestoreListener userStateRestoreListener,
                                      final StandbyUpdateListener userStandbyUpdateListener,

Review comment (suggestion):
The default should be `REMAIN_IN_GROUP`, which corresponds to the default of `leaveGroupRequested = false`

Code after:

    private final ChangelogReader changelogReader;
    private final ConsumerRebalanceListener rebalanceListener;
    private final Consumer<byte[], byte[]> mainConsumer;
    private final Consumer<byte[], byte[]> restoreConsumer;
    private final Admin adminClient;
    private final TopologyMetadata topologyMetadata;
    private final java.util.function.Consumer<Long> cacheResizer;

    private BiConsumer<Throwable, Boolean> streamsUncaughtExceptionHandler;
    private final Runnable shutdownErrorHook;

    // These must be Atomic references as they are shared and used to signal between the assignor and the stream thread
    private final AtomicInteger assignmentErrorCode;
    private final AtomicLong nextProbingRebalanceMs;
    // recoverable errors (don't require killing thread) that we need to invoke the exception
    // handler for, eg MissingSourceTopicException with named topologies
    private final Queue<StreamsException> nonFatalExceptionsToHandle;

    private final Optional<StreamsRebalanceData> streamsRebalanceData;
    private final StreamsMetadataState streamsMetadataState;

    // These are used to signal from outside the stream thread, but the variables themselves are internal to the thread
    private final AtomicLong cacheResizeSize = new AtomicLong(-1L);
    private final AtomicReference<org.apache.kafka.streams.CloseOptions.GroupMembershipOperation> leaveGroupRequested =
        new AtomicReference<>(org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP);
    private final AtomicLong lastShutdownWarningTimestamp = new AtomicLong(0L);
    private final boolean eosEnabled;
    private final boolean stateUpdaterEnabled;
    private final boolean processingThreadsEnabled;

    private volatile long fetchDeadlineClientInstanceId = -1;
    private volatile KafkaFutureImpl<Uuid> mainConsumerInstanceIdFuture = new KafkaFutureImpl<>();
    private volatile KafkaFutureImpl<Uuid> restoreConsumerInstanceIdFuture = new KafkaFutureImpl<>();
    private volatile KafkaFutureImpl<Uuid> producerInstanceIdFuture = new KafkaFutureImpl<>();

    private Timer topicsReadyTimer;

    public static StreamThread create(final TopologyMetadata topologyMetadata,
                                      final StreamsConfig config,
                                      final KafkaClientSupplier clientSupplier,
                                      final Admin adminClient,
                                      final UUID processId,
                                      final String clientId,
                                      final StreamsMetricsImpl streamsMetrics,
                                      final Time time,
                                      final StreamsMetadataState streamsMetadataState,
                                      final long cacheSizeBytes,
                                      final StateDirectory stateDirectory,
                                      final StateRestoreListener userStateRestoreListener,
                                      final StandbyUpdateListener userStandbyUpdateListener,