public void createTasks(String jobId, List<TaskAddParameter> taskList,
            Iterable<BatchClientBehavior> additionalBehaviors) throws RuntimeException, InterruptedException {

        BehaviorManager bhMgr = new BehaviorManager(this.customBehaviors(), additionalBehaviors);

        // Default thread number is 1
        int threadNumber = 1;

        // Get user defined thread number
        for (BatchClientBehavior op : bhMgr.getMasterListOfBehaviors()) {
            if (op instanceof BatchClientParallelOptions) {
                threadNumber = ((BatchClientParallelOptions) op).maxDegreeOfParallelism();
            }
        }

        final Object lock = new Object();
        ConcurrentLinkedQueue<TaskAddParameter> pendingList = new ConcurrentLinkedQueue<>(taskList);
        CopyOnWriteArrayList<TaskAddResult> failures = new CopyOnWriteArrayList<>();

        Map<Thread, WorkingThread> threads = new HashMap<>();
        Exception innerException = null;

        synchronized (lock) {
            while (!pendingList.isEmpty()) {

                if (threads.size() < threadNumber) {
                    // Kick as many as possible add tasks requests by max allowed threads
                    WorkingThread worker = new WorkingThread(this.parentBatchClient, bhMgr, jobId, pendingList,
                            failures, lock);
                    Thread thread = new Thread(worker);
                    thread.start();
                    threads.put(thread, worker);
                } else {
                    // Wait for any thread to finish
                    lock.wait();

                    List<Thread> finishedThreads = new ArrayList<>();
                    for (Map.Entry<Thread, WorkingThread> entry : threads.entrySet()) {
                        if (entry.getKey().getState() == Thread.State.TERMINATED) {
                            finishedThreads.add(entry.getKey());
                            // If any exception is encountered, then stop immediately without waiting for
                            // remaining active threads.
                            innerException = entry.getValue().getException();
                            if (innerException != null) {
                                break;
                            }
                        }
                    }

                    // Free the thread pool so we can start more threads to send the remaining add
                    // tasks requests.
                    threads.keySet().removeAll(finishedThreads);

                    // Any errors happened, we stop.
                    if (innerException != null || !failures.isEmpty()) {
                        break;
                    }
                }
            }
        }

        // Wait for all remaining threads to finish.
        for (Thread t : threads.keySet()) {
            t.join();
        }

        if (innerException == null) {
            // Check for errors in any of the threads.
            for (Map.Entry<Thread, WorkingThread> entry : threads.entrySet()) {
                innerException = entry.getValue().getException();
                if (innerException != null) {
                    break;
                }
            }
        }

        if (innerException != null) {
            // If an exception happened in any of the threads, throw it.
            if (innerException instanceof BatchErrorException) {
                throw (BatchErrorException) innerException;
            } else if (innerException instanceof  RuntimeException) {
                // WorkingThread will only catch and store a BatchErrorException or a
                // RuntimeException in its run() method.
                // WorkingThread.getException() should therefore only return one of these two
                // types, making the cast safe.
                throw (RuntimeException) innerException;
            }
        }

        if (!failures.isEmpty()) {
            // Report any client error with leftover request
            List<TaskAddParameter> notFinished = new ArrayList<>();
            for (TaskAddParameter param : pendingList) {
                notFinished.add(param);
            }
            throw new CreateTasksErrorException("At least one task failed to be added.", failures, notFinished);
        }

        // We succeed here
    }
public void createTasks ( String jobId , List < TaskAddParameter > taskList , Iterable < BatchClientBehavior > additionalBehaviors ) throws RuntimeException , InterruptedException { BehaviorManager bhMgr = new BehaviorManager ( this . customBehaviors ( ) , additionalBehaviors ) ; int threadNumber = 1 ; for ( BatchClientBehavior op : bhMgr . getMasterListOfBehaviors ( ) ) { if ( op instanceof BatchClientParallelOptions ) { threadNumber = ( ( BatchClientParallelOptions ) op ) . maxDegreeOfParallelism ( ) ; } } final Object lock = new Object ( ) ; ConcurrentLinkedQueue < TaskAddParameter > pendingList = new ConcurrentLinkedQueue < > ( taskList ) ; CopyOnWriteArrayList < TaskAddResult > failures = new CopyOnWriteArrayList < > ( ) ; Map < Thread , WorkingThread > threads = new HashMap < > ( ) ; Exception innerException = null ; synchronized ( lock ) { while ( ! pendingList . isEmpty ( ) ) { if ( threads . size ( ) < threadNumber ) { WorkingThread worker = new WorkingThread ( this . parentBatchClient , bhMgr , jobId , pendingList , failures , lock ) ; Thread thread = new Thread ( worker ) ; thread . start ( ) ; threads . put ( thread , worker ) ; } else { lock . wait ( ) ; List < Thread > finishedThreads = new ArrayList < > ( ) ; for ( Map . Entry < Thread , WorkingThread > entry : threads . entrySet ( ) ) { if ( entry . getKey ( ) . getState ( ) == Thread . State . TERMINATED ) { finishedThreads . add ( entry . getKey ( ) ) ; innerException = entry . getValue ( ) . getException ( ) ; if ( innerException != null ) { break ; } } } threads . keySet ( ) . removeAll ( finishedThreads ) ; if ( innerException != null || ! failures . isEmpty ( ) ) { break ; } } } } for ( Thread t : threads . keySet ( ) ) { t . join ( ) ; } if ( innerException == null ) { for ( Map . Entry < Thread , WorkingThread > entry : threads . entrySet ( ) ) { innerException = entry . getValue ( ) . getException ( ) ; if ( innerException != null ) { break ; } } } if ( innerException != null ) { if ( innerException instanceof BatchErrorException ) { throw ( BatchErrorException ) innerException ; } else if ( innerException instanceof RuntimeException ) { throw ( RuntimeException ) innerException ; } } if ( ! failures . isEmpty ( ) ) { List < TaskAddParameter > notFinished = new ArrayList < > ( ) ; for ( TaskAddParameter param : pendingList ) { notFinished . add ( param ) ; } throw new CreateTasksErrorException ( "At least one task failed to be added." , failures , notFinished ) ; } }
Adds multiple tasks to a job.

@param jobId
The ID of the job to which to add the task.
@param taskList
A list of {@link TaskAddParameter tasks} to add.
@param additionalBehaviors
A collection of {@link BatchClientBehavior} instances that are
applied to the Batch service request.
@throws RuntimeException
Exception thrown when an error response is received from the
Batch service or any network exception.
@throws InterruptedException
Exception thrown if any thread has interrupted the current
thread.