@Override
    public Optional<Message<T>> getMessage() {
        return sourceRecord.getMessage();
    }
@Test
    public void testCustomAck() {

        PulsarRecord pulsarRecord = Mockito.mock(PulsarRecord.class);
        SinkRecord sinkRecord = new SinkRecord<>(pulsarRecord, new Object());

        sinkRecord.cumulativeAck();
        Mockito.verify(pulsarRecord, Mockito.times(1)).cumulativeAck();

        sinkRecord = new SinkRecord(Mockito.mock(Record.class), new Object());
        try {
            sinkRecord.individualAck();
            fail("Should throw runtime exception");
        } catch (Exception e) {
            assertTrue(e instanceof RuntimeException);
            assertEquals(e.getMessage(), "SourceRecord class type must be PulsarRecord");
        }
    }