class BatchEphemeralJob extends EphemeralJob<Object> {

        private static final long PAGE_SIZE = Query.COMMIT_BATCH;
        private static final String BATCH_EPHEMERAL_JOB = "batch-ephemeral-job";
        private static final long MAX_CONSUME_COUNT = 2 * PAGE_SIZE;

        private WeakReference<EphemeralJobQueue> queueWeakReference;

        public BatchEphemeralJob(EphemeralJobQueue queue) {
            this.queueWeakReference = new WeakReference<>(queue);
        }

        @Override
        public String type() {
            return BATCH_EPHEMERAL_JOB;
        }

        @Override
        public Object execute() throws Exception {<FILL_FUNCTION_BODY>}

        private Object executeBatchJob(List<EphemeralJob<?>> jobs, Object prevResult) throws
                                                                                      Exception {
            GraphTransaction graphTx = this.params().systemTransaction();
            GraphTransaction systemTx = this.params().graphTransaction();
            Object result = prevResult;
            for (EphemeralJob<?> job : jobs) {
                this.initJob(job);
                Object obj = job.call();
                if (job instanceof Reduce) {
                    result = ((Reduce) job).reduce(result, obj);
                }
            }

            graphTx.commit();
            systemTx.commit();

            return result;
        }

        private void initJob(EphemeralJob<?> job) {
            job.graph(this.graph());
            job.params(this.params());
        }

        @Override
        public Object call() throws Exception {
            try {
                return super.call();
            } catch (Throwable e) {
                LOG.warn("Failed to execute BatchEphemeralJob", e);
                EphemeralJobQueue queue = this.queueWeakReference.get();
                if (e instanceof InterruptedException) {
                    Thread.currentThread().interrupt();
                    if (queue != null) {
                        queue.clear();
                        queue.consumeComplete();
                    }
                    throw e;
                }

                if (queue != null) {
                    queue.consumeComplete();
                    if (!queue.isEmpty()) {
                        queue.reScheduleIfNeeded();
                    }
                }
                throw e;
            }
        }
    }

            boolean stop = false;
            Object result = null;
            int consumeCount = 0;
            InterruptedException interruptedException = null;
            EphemeralJobQueue queue;
            List<EphemeralJob<?>> batchJobs = new ArrayList<>();
            while (!stop) {
                if (interruptedException == null && Thread.currentThread().isInterrupted()) {
                    interruptedException = new InterruptedException();
                }

                queue = this.queueWeakReference.get();
                if (queue == null) {
                    stop = true;
                    continue;
                }

                if (queue.isEmpty() || consumeCount > MAX_CONSUME_COUNT ||
                    interruptedException != null) {
                    queue.consumeComplete();
                    stop = true;
                    if (!queue.isEmpty()) {
                        queue.reScheduleIfNeeded();
                    }
                    continue;
                }

                try {
                    while (!queue.isEmpty() && batchJobs.size() < PAGE_SIZE) {
                        EphemeralJob<?> job = queue.poll();
                        if (job == null) {
                            continue;
                        }
                        batchJobs.add(job);
                    }

                    if (batchJobs.isEmpty()) {
                        continue;
                    }

                    consumeCount += batchJobs.size();
                    result = this.executeBatchJob(batchJobs, result);

                } catch (InterruptedException e) {
                    interruptedException = e;
                } finally {
                    batchJobs.clear();
                }
            }

            if (interruptedException != null) {
                Thread.currentThread().interrupt();
                throw interruptedException;
            }

            return result;
        
<no_super_class>