/*
 * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. 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.wso2.andes.kernel.slot;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.wso2.andes.kernel.AndesException;
import org.wso2.andes.kernel.MessagingEngine;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

/**
 * This class is responsible for deleting slots and scheduling slot deletions.
 */
public class SlotDeletionExecutor {

    private static Log log = LogFactory.getLog(SlotDeletionExecutor.class);


    private LinkedBlockingQueue<Slot> slotsToDelete = new LinkedBlockingQueue<Slot>();

    /**
     * Slot deletion thread factory in one MB node
     */
    private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat
            ("SlotDeletionExecutor-%d").build();

    /**
     * Slot deletion executor service
     */
    private ExecutorService slotDeletionExecutorService;

    /**
     * SlotDeletionScheduler instance
     */
    private static SlotDeletionExecutor instance;

    /**
     * SlotDeletionExecutor constructor
     */
    private SlotDeletionExecutor() {

    }

    /**
     * Create slot deletion scheduler
     */
    public void init() {
        this.slotDeletionExecutorService = Executors.newSingleThreadExecutor(namedThreadFactory);
        this.slotDeletionExecutorService.submit(new SlotDeletionTask());

    }

    /**
     * Slot deletion task running and take slot from queue.
     */
    class SlotDeletionTask implements Runnable {

        //Slot which previously attempt to delete
        Slot previouslyAttemptedSlot = null;

        /**
         * Running slot deletion task
         */
        public void run() {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    // Slot to attempt current deletion
                    Slot slot = slotsToDelete.poll(1, TimeUnit.SECONDS);

                    // Check current slot to delete is not null
                    if (slot != null) {

                        // Check DB for any remaining messages. (JIRA FIX: MB-1612)
                        // If there are any remaining messages wait till overlapped slot delivers the messages
                        if (MessagingEngine.getInstance().getMessageCountForQueueInRange(
                                slot.getStorageQueueName(), slot.getStartMessageId(), slot.getEndMessageId()) == 0) {
                            // Invoke coordinator to delete slot
                            boolean deleteSuccess = deleteSlotAtCoordinator(slot);
                            if (!deleteSuccess) {
                                // Delete attempt not success, therefore adding slot to the queue
                                slotsToDelete.put(slot);
                            } else {
                                SlotDeliveryWorker slotWorker = SlotDeliveryWorkerManager.getInstance()
                                        .getSlotWorker(slot.getStorageQueueName());
                                slotWorker.deleteSlot(slot);
                            }
                        } else {
                            slotsToDelete.put(slot); // Not deleted. Hence puttng back in queue
                        }
                    }

                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.error("Error occurred while trying to delete the slot.", e);
                } catch (AndesException e) {
                    log.error("Error occurred while trying to delete the slot.", e);
                }
            }
        }

        /**
         * Delete slot at coordinator and return delete status
         *
         * @param slot slot to be removed from cluster
         * @return slot deletion status
         */
        private boolean deleteSlotAtCoordinator(Slot slot) {
            boolean deleteSuccess = false;
            try {
                deleteSuccess = MessagingEngine.getInstance().getSlotCoordinator().deleteSlot
                        (slot.getStorageQueueName(), slot);
            } catch (ConnectionException e) {
                log.error("Error while trying to delete the slot " + slot + " Thrift connection failed. " +
                        "Rescheduling delete.");

            }
            return deleteSuccess;
        }
    }

    /**
     * Schedule a slot for deletion. This will continuously try to delete the slot
     * until slot manager informs that the slot is successfully removed
     *
     * @param slot slot to be removed from cluster
     */
    public void executeSlotDeletion(Slot slot) {
        slotsToDelete.add(slot);

    }

    /**
     * Shutdown slot deletion executor service
     */
    public void stopSlotDeletionExecutor() {
        if (slotDeletionExecutorService != null) {
            slotDeletionExecutorService.shutdown();
        }
    }

    /**
     * Return slot deletion scheduler object
     *
     * @return SlotDeletionScheduler object
     */
    public static SlotDeletionExecutor getInstance() {
        if (instance == null) {
            instance = new SlotDeletionExecutor();
        }
        return instance;
    }

}

@@ -1875,24 +1875,127 @@
  instance;%0A%0A
+    /**%0A     * Reference of the deletion task.%0A     */%0A    private SlotDeletionTask slotDeletionTask;%0A%0A
     /**%0A    
@@ -2260,16 +2260,67 @@
 ctory);%0A
+        slotDeletionTask = new SlotDeletionTask();%0A
         
@@ -2359,21 +2359,17 @@
 .submit(
-new S
+s
 lotDelet
@@ -2379,14 +2379,11 @@
 Task
-()
 );%0A
-%0A
     
@@ -2524,93 +2524,167 @@
    /
-/Slot which previously attempt to delete%0A        Slot previouslyAttemptedSlot = null;
+**%0A         * Condition running the task.%0A         */%0A        boolean isLive = true;%0A%0A        void setLive(boolean live) %7B%0A            isLive = live;%0A        %7D
 %0A%0A  
@@ -2794,47 +2794,14 @@
 le (
-!Thread.currentThread().isInterrupted()
+isLive
 ) %7B%0A
@@ -4441,36 +4441,54 @@
  log.error(%22
-Error occurr
+SlotDeletionTask was interrupt
 ed while try
@@ -4546,26 +4546,28 @@
 ch (
-AndesException 
+Throwable throwabl
 e)
- 
 %7B%0A  
@@ -4591,25 +4591,36 @@
  log.error(%22
-E
+Unexpected e
 rror occurre
@@ -4648,32 +4648,40 @@
 ete the slot.%22, 
+throwabl
 e);%0A            
@@ -4692,32 +4692,142 @@
 %7D%0A            %7D%0A
+            log.info(%22SlotDeletionExecutor has shutdown with %22 + slotsToDelete.size() + %22 slots to delete.%22);%0A
         %7D%0A%0A     
@@ -6029,32 +6029,77 @@
 vice != null) %7B%0A
+            slotDeletionTask.setLive(false);%0A
             slot

Fix SlotDeletionTask while loop and adding error logs for context.