AbstractJobDefine job5 =  
        new FailoverConsumerTest(event, "故障转移消费测试", pulsarClient, 20, admin);  
CompletableFuture<Void> c5 = CompletableFuture.runAsync(job5::start, EXECUTOR);  
AbstractJobDefine job6 = new SchemaTest(event,"schema测试",pulsarClient,20,prestoService);  
CompletableFuture<Void> c6 = CompletableFuture.runAsync(job6::start, EXECUTOR);  
AbstractJobDefine job7 = new VlogsTest(event,"vlogs test",pulsarClient,20, vlogsUrl);  
CompletableFuture<Void> c7 = CompletableFuture.runAsync(job7::start, EXECUTOR);  
  
CompletableFuture<Void> all = CompletableFuture.allOf(c1, c2, c3, c4, c5, c6, c7);  
all.whenComplete((___, __) -> {  
    event.finishAll();  
    pulsarClient.closeAsync();  
    admin.close();  
}).get();

@Data
public abstract class AbstractJobDefine {
    private Event event;
    private String jobName;
    private PulsarClient pulsarClient;

    private int timeout;

    private PulsarAdmin admin;

    public AbstractJobDefine(Event event, String jobName, PulsarClient pulsarClient, int timeout, PulsarAdmin admin) {
        this.event = event;
        this.jobName = jobName;
        this.pulsarClient = pulsarClient;
        this.timeout = timeout;
        this.admin = admin;
    }

    public void start() {
        event.addJob();
        try {
            CompletableFuture.runAsync(() -> {
                StopWatch watch = new StopWatch();
                try {
                    watch.start(jobName);
                    run(pulsarClient, admin);
                } catch (Exception e) {
                    event.oneException(this, e);
                } finally {
                    watch.stop();
                    event.finishOne(jobName, StrUtil.format("cost: {}s", watch.getTotalTimeSeconds()));
                }
            }, TestCase.EXECUTOR).get(timeout, TimeUnit.SECONDS);
        } catch (Exception e) {
            event.oneException(this, e);
        }
    }


    /** run busy code
     * @param pulsarClient pulsar client
     * @param admin pulsar admin client
     * @throws Exception e
     */
    public abstract void run(PulsarClient pulsarClient, PulsarAdmin admin) throws Exception;
}

public interface Event {  
  
    /**  
     * 新增一个任务  
     */  
    void addJob();  
  
    /** 获取运行中的任务数量  
     * @return 获取运行中的任务数量  
     */  
    TestCaseRuntimeResponse getRuntime();  
  
    /**  
     * 单个任务执行完毕  
     *  
     * @param jobName    任务名称  
     * @param finishCost 任务完成耗时  
     */  
    void finishOne(String jobName, String finishCost);  
  
    /**单个任务执行异常  
     * @param jobDefine 任务  
     * @param e 异常  
     */  
    void oneException(AbstractJobDefine jobDefine, Exception e);  
  
    /**  
     * 所有任务执行完毕  
     */  
    void finishAll();  
}