/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.kafka.coordinator.group.classic;

import java.util.Arrays;
import java.util.HashSet;
import java.util.Locale;
import java.util.Set;

/**
 * Represents all states that a classic group can be in, as well as the states that a group must
 * be in to transition to a particular state.
 */
public enum ClassicGroupState {

    /**
     * Group has no more members, but lingers until all offsets have expired. This state
     * also represents groups which use Kafka only for offset commits and have no members.
     *
     * action: respond normally to join group from new members
     *         respond to sync group with UNKNOWN_MEMBER_ID
     *         respond to heartbeat with UNKNOWN_MEMBER_ID
     *         respond to leave group with UNKNOWN_MEMBER_ID
     *         respond to offset commit with UNKNOWN_MEMBER_ID
     *         allow offset fetch requests
     * transition: last offsets removed in periodic expiration task => DEAD
     *             join group from a new member => PREPARING_REBALANCE
     *             group is removed by partition emigration => DEAD
     *             group is removed by expiration => DEAD
     */
    EMPTY("Empty"),

    /**
     * Group is preparing to rebalance.
     *
     * action: respond to heartbeats with REBALANCE_IN_PROGRESS
     *         respond to sync group with REBALANCE_IN_PROGRESS
     *         remove member on leave group request
     *         park join group requests from new or existing members until all expected members have joined
     *         allow offset commits from previous generation
     *         allow offset fetch requests
     * transition: some members have joined by the timeout => COMPLETING_REBALANCE
     *             all members have left the group => EMPTY
     *             group is removed by partition emigration => DEAD
     */
    PREPARING_REBALANCE("PreparingRebalance"),

    /**
     * Group is awaiting state assignment from the leader.
     *
     * action: respond to heartbeats with REBALANCE_IN_PROGRESS
     *         respond to offset commits with REBALANCE_IN_PROGRESS
     *         park sync group requests from followers until transition to STABLE
     *         allow offset fetch requests
     * transition: sync group with state assignment received from leader => STABLE
     *             join group from new member or existing member with updated metadata => PREPARING_REBALANCE
     *             leave group from existing member => PREPARING_REBALANCE
     *             member failure detected => PREPARING_REBALANCE
     *             group is removed by partition emigration => DEAD
     */
    COMPLETING_REBALANCE("CompletingRebalance"),

    /**
     * Group is stable.
     *
     * action: respond to member heartbeats normally
     *         respond to sync group from any member with current assignment
     *         respond to join group from followers with matching metadata with current group metadata
     *         allow offset commits from member of current generation
     *         allow offset fetch requests
     * transition: member failure detected via heartbeat => PREPARING_REBALANCE
     *             leave group from existing member => PREPARING_REBALANCE
     *             leader join-group received => PREPARING_REBALANCE
     *             follower join-group with new metadata => PREPARING_REBALANCE
     *             group is removed by partition emigration => DEAD
     */
    STABLE("Stable"),

    /**
     * Group has no more members and its metadata is being removed.
     *
     * action: respond to join group with UNKNOWN_MEMBER_ID
     *         respond to sync group with UNKNOWN_MEMBER_ID
     *         respond to heartbeat with UNKNOWN_MEMBER_ID
     *         respond to leave group with UNKNOWN_MEMBER_ID
     *         respond to offset commit with UNKNOWN_MEMBER_ID
     *         allow offset fetch requests
     * transition: DEAD is a final state before group metadata is cleaned up, so there are no transitions
     */
    DEAD("Dead");

    private final String name;
    private final String lowerCaseName;
    private Set<ClassicGroupState> validPreviousStates;

    static {
        EMPTY.addValidPreviousStates(PREPARING_REBALANCE);
        PREPARING_REBALANCE.addValidPreviousStates(STABLE, COMPLETING_REBALANCE, EMPTY);
        COMPLETING_REBALANCE.addValidPreviousStates(PREPARING_REBALANCE);
        STABLE.addValidPreviousStates(COMPLETING_REBALANCE);
        DEAD.addValidPreviousStates(STABLE, PREPARING_REBALANCE, COMPLETING_REBALANCE, EMPTY, DEAD);
    }

    ClassicGroupState(String name) {
        this.name = name;
        this.lowerCaseName = name.toLowerCase(Locale.ROOT);
    }

    @Override
    public String toString() {
        return name;
    }

    public String toLowerCaseString() {
        return lowerCaseName;
    }

    private void addValidPreviousStates(ClassicGroupState... validPreviousStates) {
        this.validPreviousStates = new HashSet<>(Arrays.asList(validPreviousStates));
    }

    /**
     * @return valid previous states a group must be in to transition to this state.
     */
    public Set<ClassicGroupState> validPreviousStates() {
        return this.validPreviousStates;
    }
}
