Repository: apache/flink
File: flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxImpl.java (Java)

Code before:
        if (head != null) {
            return Optional.of(head);
        }
        if (!hasNewMail) {
            return Optional.empty();
        }
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            final Mail value = takeOrNull(queue, priority);
            if (value == null) {
                return Optional.empty();
            }
            correctHasNewMail();
            return Optional.ofNullable(value);
        } finally {
            lock.unlock();
        }
    }

    @Override
    public @Nonnull Mail take(int priority) throws InterruptedException, IllegalStateException {
        checkIsMailboxThread();
        checkTakeStateConditions();

        moveMailsToBatchIfNeeded();

        Mail head = takeOrNull(batch, priority);
        if (head != null) {
            return head;
        }
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            Mail headMail;
            while ((headMail = takeOrNull(queue, priority)) == null) {
                // to ease debugging
                notEmpty.await(1, TimeUnit.SECONDS);
            }
            correctHasNewMail();
            return headMail;
        } finally {
            lock.unlock();
        }
    }

    private void correctHasNewMail() {
        Mail peek = queue.peek();
        if (peek != null) {
            hasNewMail = true;
            hasNewHighPriorityMail = peek.getMailOptions().isHighPriority();

Review comment (performance):
I see we are creating batches more aggressively now whenever we get high priority mails. That would cause increased locking for all control calls. Can it have performance implications?

Code after:
        if (head != null) {
            return Optional.of(head);
        }
        if (!hasNewMail) {
            return Optional.empty();
        }
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            final Mail value = takeOrNull(queue, priority);
            if (value == null) {
                return Optional.empty();
            }
            updateNewMailFlags();
            return Optional.ofNullable(value);
        } finally {
            lock.unlock();
        }
    }

    @Override
    public @Nonnull Mail take(int priority) throws InterruptedException, IllegalStateException {
        checkIsMailboxThread();
        checkTakeStateConditions();

        moveUrgentMailsToBatchIfNeeded(true);

        Mail head = takeOrNull(batch, priority);
        if (head != null) {
            return head;
        }
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            Mail headMail;
            while ((headMail = takeOrNull(queue, priority)) == null) {
                // to ease debugging
                notEmpty.await(1, TimeUnit.SECONDS);
            }
            updateNewMailFlags();
            return headMail;
        } finally {
            lock.unlock();
        }
    }

    private void updateNewMailFlags() {
        Mail peek = queue.peek();
        if (peek != null) {
            hasNewMail = true;
            hasNewUrgentMail = peek.getMailOptions().isUrgent();