@VisibleForTesting
    void handleResult(Record srcRecord, JavaExecutionResult result) throws Exception {
        if (result.getUserException() != null) {
            Exception t = result.getUserException();
            log.warn("Encountered exception when processing message {}",
                    srcRecord, t);
            stats.incrUserExceptions(t);
            srcRecord.fail();
        } else {
            if (result.getResult() != null) {
                sendOutputMessage(srcRecord, result.getResult());
            } else {
                org.apache.pulsar.functions.proto.Function.FunctionDetails functionDetails =
                        instanceConfig.getFunctionDetails();
                // When function return null, needs to be acked directly.
                if (functionDetails.getProcessingGuarantees()
                        != org.apache.pulsar.functions.proto.Function.ProcessingGuarantees.MANUAL) {
                    // This condition has been automatically acked.
                    // After waiting to remove the autoAck configuration,can be removing the judgment condition.
                    if (!functionDetails.getAutoAck() || functionDetails.getProcessingGuarantees()
                            != org.apache.pulsar.functions.proto.Function.ProcessingGuarantees.ATMOST_ONCE) {
                        srcRecord.ack();
                    }
                }
            }
            // increment total successfully processed
            stats.incrTotalProcessedSuccessfully();
        }
    }
@Test
    public void testFunctionResultNull() throws Exception {
        JavaExecutionResult javaExecutionResult = new JavaExecutionResult();

        // ProcessingGuarantees == MANUAL, not need ack.
        Record record = mock(Record.class);
        getJavaInstanceRunnable(true, org.apache.pulsar.functions.proto.Function.ProcessingGuarantees.MANUAL)
                .handleResult(record, javaExecutionResult);
        verify(record, times(0)).ack();

        // ProcessingGuarantees == ATMOST_ONCE and autoAck == true, not need ack
        clearInvocations(record);
        getJavaInstanceRunnable(true, org.apache.pulsar.functions.proto.Function.ProcessingGuarantees.ATMOST_ONCE)
                .handleResult(record, javaExecutionResult);
        verify(record, times(0)).ack();

        // other case, need ack
        clearInvocations(record);
        getJavaInstanceRunnable(true, org.apache.pulsar.functions.proto.Function.ProcessingGuarantees.ATLEAST_ONCE)
                .handleResult(record, javaExecutionResult);
        verify(record, times(1)).ack();
        clearInvocations(record);
        getJavaInstanceRunnable(true, org.apache.pulsar.functions.proto.Function.ProcessingGuarantees.EFFECTIVELY_ONCE)
                .handleResult(record, javaExecutionResult);
        verify(record, times(1)).ack();
    }