@VisibleForTesting
    static BatchMessageSequenceRef getMessageSequenceRefForBatchMessage(MessageId messageId) {
        long ledgerId;
        long entryId;
        int batchIdx;
        try {
            try {
                batchIdx = (int) getMethodOfMessageId(messageId, "getBatchIndex").invoke(messageId);
                if (batchIdx < 0) {
                    return null;
                }
            } catch (NoSuchMethodException noSuchMethodException) {
                // not a BatchMessageIdImpl, returning null to use the standard sequenceId
                return null;
            }

            ledgerId = (long) getMethodOfMessageId(messageId, "getLedgerId").invoke(messageId);
            entryId = (long) getMethodOfMessageId(messageId, "getEntryId").invoke(messageId);
        } catch (IllegalAccessException | NoSuchMethodException | InvocationTargetException ex) {
            log.error("Unexpected error while retrieving sequenceId, messageId class: {}, error: {}",
                    messageId.getClass().getName(), ex.getMessage(), ex);
            throw new RuntimeException(ex);
        }

        return new BatchMessageSequenceRef(ledgerId, entryId, batchIdx);
    }
@Test
    public void testGetMessageSequenceRefForBatchMessage() throws Exception {
        long ledgerId = 123L;
        long entryId = Long.MAX_VALUE;
        int batchIdx = 16;

        KafkaConnectSink.BatchMessageSequenceRef ref = KafkaConnectSink
                .getMessageSequenceRefForBatchMessage(new MessageIdImpl(ledgerId, entryId, 0));
        assertNull(ref);

        ref = KafkaConnectSink.getMessageSequenceRefForBatchMessage(
                        new TopicMessageIdImpl("topic-0", new MessageIdImpl(ledgerId, entryId, 0))
        );
        assertNull(ref);

        ref = KafkaConnectSink.getMessageSequenceRefForBatchMessage(
                new BatchMessageIdImpl(ledgerId, entryId, 0, batchIdx));

        assertEquals(ref.getLedgerId(), ledgerId);
        assertEquals(ref.getEntryId(), entryId);
        assertEquals(ref.getBatchIdx(), batchIdx);

        ref = KafkaConnectSink.getMessageSequenceRefForBatchMessage(
                new TopicMessageIdImpl("topic-0", new BatchMessageIdImpl(ledgerId, entryId, 0, batchIdx))
        );

        assertEquals(ref.getLedgerId(), ledgerId);
        assertEquals(ref.getEntryId(), entryId);
        assertEquals(ref.getBatchIdx(), batchIdx);
    }