public static boolean isEntryExpired(int messageTTLInSeconds, long entryTimestamp) {
        return messageTTLInSeconds != 0
                && (System.currentTimeMillis() > entryTimestamp + TimeUnit.SECONDS.toMillis(messageTTLInSeconds));
    }
@Test(timeOut = 30000)
    public void testMessageBrokerAndEntryMetadataTimestampMissed() {
        int MOCK_BATCH_SIZE = 10;
        String data = "test-message";
        ByteBuf byteBuf = PulsarByteBufAllocator.DEFAULT.buffer(data.length(), data.length());
        byteBuf.writeBytes(data.getBytes(StandardCharsets.UTF_8));

        try {
            // test BrokerTimestamp not set.
            MessageMetadata messageMetadata = new MessageMetadata()
                    .setPublishTime(1)
                    .setProducerName("test")
                    .setSequenceId(1);
            byteBuf = Commands.serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, messageMetadata, byteBuf);
            BrokerEntryMetadata brokerMetadata = new BrokerEntryMetadata()
                            .setIndex(MOCK_BATCH_SIZE - 1);

            int brokerMetaSize = brokerMetadata.getSerializedSize();
            ByteBuf  brokerMeta = PulsarByteBufAllocator.DEFAULT.buffer(brokerMetaSize + 6, brokerMetaSize + 6);
            brokerMeta.writeShort(Commands.magicBrokerEntryMetadata);
            brokerMeta.writeInt(brokerMetaSize);
            brokerMetadata.writeTo(brokerMeta);

            CompositeByteBuf compositeByteBuf = PulsarByteBufAllocator.DEFAULT.compositeBuffer();
            compositeByteBuf.addComponents(true, brokerMeta, byteBuf);
            long entryTimestamp = Commands.getEntryTimestamp(compositeByteBuf);
            assertTrue(MessageImpl.isEntryExpired(100, entryTimestamp));
            assertEquals(entryTimestamp, 1);

            // test BrokerTimestamp set.
            byteBuf = PulsarByteBufAllocator.DEFAULT.buffer(data.length(), data.length());
            byteBuf.writeBytes(data.getBytes(StandardCharsets.UTF_8));
            messageMetadata = new MessageMetadata()
                    .setPublishTime(System.currentTimeMillis())
                    .setProducerName("test")
                    .setSequenceId(1);
            byteBuf = Commands.serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, messageMetadata, byteBuf);
            long brokerEntryTimestamp = System.currentTimeMillis();
            brokerMetadata = new BrokerEntryMetadata()
                    .setBrokerTimestamp(brokerEntryTimestamp)
                    .setIndex(MOCK_BATCH_SIZE - 1);

            brokerMetaSize = brokerMetadata.getSerializedSize();
            brokerMeta = PulsarByteBufAllocator.DEFAULT.buffer(brokerMetaSize + 6, brokerMetaSize + 6);
            brokerMeta.writeShort(Commands.magicBrokerEntryMetadata);
            brokerMeta.writeInt(brokerMetaSize);
            brokerMetadata.writeTo(brokerMeta);

            compositeByteBuf = PulsarByteBufAllocator.DEFAULT.compositeBuffer();
            compositeByteBuf.addComponents(true, brokerMeta, byteBuf);
            entryTimestamp = Commands.getEntryTimestamp(compositeByteBuf);
            assertFalse(MessageImpl.isEntryExpired(24 * 3600, entryTimestamp));
            assertEquals(entryTimestamp, brokerEntryTimestamp);
        } catch (IOException e) {
            fail();
        }
    }