public boolean containsMessage(MessageExt message) {
        if (message == null) {
            // should never reach here.
            return false;
        }
        try {
            this.treeMapLock.readLock().lockInterruptibly();
            try {
                return this.msgTreeMap.containsKey(message.getQueueOffset());
            } finally {
                this.treeMapLock.readLock().unlock();
            }
        } catch (Throwable t) {
            log.error("Failed to check message's existence in process queue, message={}", message, t);
        }
        return false;
    }
@Test
    public void testContainsMessage() {
        ProcessQueue pq = new ProcessQueue();
        final List<MessageExt> messageList = createMessageList(2);
        final MessageExt message0 = messageList.get(0);
        final MessageExt message1 = messageList.get(1);

        pq.putMessage(Lists.list(message0));
        assertThat(pq.containsMessage(message0)).isTrue();
        assertThat(pq.containsMessage(message1)).isFalse();
    }