protected Record<T> buildRecord(Consumer<T> consumer, Message<T> message) {
        Schema<T> schema = null;
        if (message instanceof MessageImpl) {
            MessageImpl impl = (MessageImpl) message;
            schema = impl.getSchemaInternal();
        } else if (message instanceof TopicMessageImpl) {
            TopicMessageImpl impl = (TopicMessageImpl) message;
            schema = impl.getSchemaInternal();
        }

        // we don't want the Function/Sink to see AutoConsumeSchema
        if (schema instanceof AutoConsumeSchema) {
            AutoConsumeSchema autoConsumeSchema = (AutoConsumeSchema) schema;
            // we cannot use atSchemaVersion, because atSchemaVersion is only
            // able to decode data, here we want a Schema that
            // is able to re-encode the payload when needed.
            schema = (Schema<T>) autoConsumeSchema
                    .unwrapInternalSchema(message.getSchemaVersion());
        }
        return PulsarRecord.<T>builder()
                .message(message)
                .schema(schema)
                .topicName(message.getTopicName())
                .customAckFunction(cumulative -> {
                    if (cumulative) {
                        consumer.acknowledgeCumulativeAsync(message)
                                .whenComplete((unused, throwable) -> message.release());
                    } else {
                        consumer.acknowledgeAsync(message).whenComplete((unused, throwable) -> message.release());
                    }
                })
                .ackFunction(() -> {
                    if (pulsarSourceConfig
                            .getProcessingGuarantees() == FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE) {
                        consumer.acknowledgeCumulativeAsync(message)
                                .whenComplete((unused, throwable) -> message.release());
                    } else {
                        consumer.acknowledgeAsync(message).whenComplete((unused, throwable) -> message.release());
                    }
                }).failFunction(() -> {
                    try {
                        if (pulsarSourceConfig.getProcessingGuarantees()
                                == FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE) {
                            throw new RuntimeException("Failed to process message: " + message.getMessageId());
                        }
                        consumer.negativeAcknowledge(message);
                    } finally {
                        // don't need to check if message pooling is set
                        // client will automatically check
                        message.release();
                    }
                })
                .build();
    }
@Test(dataProvider = "sourceImpls")
    public void testPulsarRecordCustomAck(PulsarSourceConfig pulsarSourceConfig) throws Exception {

        PulsarSource pulsarSource = getPulsarSource(pulsarSourceConfig);
        Message message = Mockito.mock(Message.class);
        Consumer consumer = Mockito.mock(Consumer.class);
        Mockito.when(consumer.acknowledgeAsync(message)).thenReturn(CompletableFuture.completedFuture(null));
        Mockito.when(consumer.acknowledgeCumulativeAsync(message)).thenReturn(CompletableFuture.completedFuture(null));

        PulsarRecord record = (PulsarRecord) pulsarSource.buildRecord(consumer, message);

        record.cumulativeAck();
        Mockito.verify(consumer, Mockito.times(1)).acknowledgeCumulativeAsync(message);

        record.individualAck();
        Mockito.verify(consumer, Mockito.times(1)).acknowledgeAsync(message);
    }