using FSM for kafka consumer