DataCenterCDC-DS/escheduler-common/src/test/java/cn/escheduler/common/queue/TaskQueueImplTest.java

/*
 * 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 cn.escheduler.common.queue;

import cn.escheduler.common.Constants;
import cn.escheduler.common.utils.IpUtils;
import cn.escheduler.common.utils.OSUtils;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.List;
import java.util.Random;

import static org.junit.Assert.assertEquals;

/**
 * task queue test
 */
public class TaskQueueImplTest {

    private static final Logger logger = LoggerFactory.getLogger(TaskQueueImplTest.class);

    ITaskQueue tasksQueue = null;

    @Before
    public void before(){
        tasksQueue = TaskQueueFactory.getTaskQueueInstance();
        //clear all data
        tasksQueue.delete();

    }


    @After
    public void after(){
        //clear all data
        tasksQueue.delete();
    }


    @Test
    public void testAdd(){


        //add
        tasksQueue.add(Constants.SCHEDULER_TASKS_QUEUE,"1_0_1_1_-1");
        tasksQueue.add(Constants.SCHEDULER_TASKS_QUEUE,"0_1_1_1_2130706433,3232236775");
        tasksQueue.add(Constants.SCHEDULER_TASKS_QUEUE,"1_1_0_1_2130706433,3232236775,"+IpUtils.ipToLong(OSUtils.getHost()));
        tasksQueue.add(Constants.SCHEDULER_TASKS_QUEUE,"1_2_1_1_2130706433,3232236775");

        List<String> tasks = tasksQueue.poll(Constants.SCHEDULER_TASKS_QUEUE, 1);

        if(tasks.size() <= 0){
            return;
        }

        //pop
        String node1 = tasks.get(0);

        assertEquals(node1,"1_0_1_1_-1");

        tasks = tasksQueue.poll(Constants.SCHEDULER_TASKS_QUEUE, 1);

        if(tasks.size() <= 0){
            return;
        }

        String node2 = tasks.get(0);

    }



    /**
     * test one million data from zookeeper queue
     */
    @Test
    public void extremeTest(){
        int total = 30 * 10000;

        for(int i = 0; i < total; i++)
        {
            for(int j = 0; j < total; j++) {
                //${processInstancePriority}_${processInstanceId}_${taskInstancePriority}_${taskId}
                //format ${processInstancePriority}_${processInstanceId}_${taskInstancePriority}_${taskId}
                String formatTask = String.format("%s_%d_%s_%d", i, i + 1, j, j == 0 ?  0 : j + new Random().nextInt(100));
                tasksQueue.add(Constants.SCHEDULER_TASKS_QUEUE, formatTask);
            }
        }

        String node1 = tasksQueue.poll(Constants.SCHEDULER_TASKS_QUEUE, 1).get(0);
        assertEquals(node1,"0");

    }

}