protected void writeOneBatch() throws InterruptedException {
        // waiting if necessary until an element becomes available
        currentJournal = journalQueue.take();
        long nextJournalId = nextVisibleJournalId;
        initBatch();

        try {
            this.journal.batchWriteBegin();

            while (true) {
                journal.batchWriteAppend(nextJournalId, currentJournal.getBuffer());
                currentBatchTasks.add(currentJournal);
                nextJournalId += 1;

                if (shouldCommitNow()) {
                    break;
                }

                currentJournal = journalQueue.take();
            }
        } catch (JournalException e) {
            // abort current task
            LOG.warn("failed to write batch, will abort current journal {} and commit", currentJournal, e);
            abortJournalTask(currentJournal, e.getMessage());
        } finally {
            try {
                // commit
                journal.batchWriteCommit();
                LOG.debug("batch write commit success, from {} - {}", nextVisibleJournalId, nextJournalId);
                nextVisibleJournalId = nextJournalId;
                markCurrentBatchSucceed();
            } catch (JournalException e) {
                // abort
                LOG.warn("failed to commit batch, will abort current {} journals.",
                        currentBatchTasks.size(), e);
                try {
                    journal.batchWriteAbort();
                } catch (JournalException e2) {
                    LOG.warn("failed to abort batch, will ignore and continue.", e);
                }
                abortCurrentBatch(e.getMessage());
            }
        }

        rollJournalAfterBatch();

        updateBatchMetrics();
    }
@Ignore
    @Test
    public void testTooManyLogs() throws Exception {
        Config.metadata_journal_max_batch_cnt = 2;
        Config.metadata_journal_max_batch_size_mb = 2;
        new Expectations(journal) {
            {
                journal.batchWriteBegin();
                times = 2;
                journal.batchWriteAppend(1, (DataOutputBuffer) any);
                times = 1;
                journal.batchWriteAppend(2, (DataOutputBuffer) any);
                times = 1;
                journal.batchWriteAppend(3, (DataOutputBuffer) any);
                times = 1;
                journal.batchWriteCommit();
                times = 2;
            }
        };
        // 4 logs
        // round 1: 2 logs
        JournalTask expectConsumedEntity = new JournalTask(System.nanoTime(), makeBuffer(10), -1);
        journalQueue.add(expectConsumedEntity);
        JournalTask expectConsumedEntity2 = new JournalTask(System.nanoTime(), makeBuffer(10), -1);
        journalQueue.add(expectConsumedEntity2);
        // round 2: 1 big log
        JournalTask bigLog = new JournalTask(System.nanoTime(), makeBuffer(2 * 1024 * 1024 - 8), -1);
        Assert.assertEquals(2 * 1024 * 1024, bigLog.estimatedSizeByte());
        journalQueue.add(bigLog);
        // this one should be left in queue
        JournalTask expectNotConsumedEntity = new JournalTask(System.nanoTime(), makeBuffer(10), -1);
        journalQueue.add(expectNotConsumedEntity);

        // round 1
        Assert.assertEquals(1, writer.nextVisibleJournalId);
        writer.writeOneBatch();
        Assert.assertEquals(3, writer.nextVisibleJournalId);
        Assert.assertEquals(2, writer.rollJournalCounter);
        Assert.assertEquals(2, writer.currentBatchTasks.size());
        Assert.assertEquals(0, expectConsumedEntity.latch.getCount());
        Assert.assertEquals(0, expectConsumedEntity2.latch.getCount());
        // two log left
        Assert.assertEquals(1, bigLog.latch.getCount());
        Assert.assertEquals(1, expectNotConsumedEntity.latch.getCount());
        Assert.assertEquals(2, journalQueue.size());

        // round 2
        writer.writeOneBatch();
        Assert.assertEquals(4, writer.nextVisibleJournalId);
        Assert.assertEquals(3, writer.rollJournalCounter);
        Assert.assertEquals(1, writer.currentBatchTasks.size());
        Assert.assertEquals(0, bigLog.latch.getCount());
        // 1 log left
        Assert.assertEquals(1, expectNotConsumedEntity.latch.getCount());
        Assert.assertEquals(1, journalQueue.size());
    }