PIP-23: Pulsar Java Client Interceptors.

[codelipenghui]

Motivation

Support user to add interceptors to producer and consumer.

Modifications

Add Consumer interceptors.

Message<T> beforeConsume(Message<T> message);
void onAcknowledge(MessageId messageId, Throwable cause);
void onAcknowledgeCumulative(MessageId messageId, Throwable cause);

Add Producer interceptors.

Message<T> beforeSend(Message<T> message);
void onSendAcknowledgement(Message<T> message, MessageId msgId, Throwable cause);

Result

Users can using interceptors in multiple scenarios, such as for applications to add
custom logging or processing.

Master Issue: #2476

[sijie]

@codelipenghui the change LGTM! a few nits around using mockito

[merlimat]

The change LGTM, just left few comments.

One thing we should consider is to add ReaderInterceptor as well. Even though the implementation is completely based on consumer, from an API perspective they are completely different.

Internally the ReaderInterceptor would be just a wrapper of ConsumerInterceptor.

Another idea, which might be out of the scope of this PR is to have Client level interceptor, both to be automatically attached to any producer/consumer created by the client instance and to also intercept events like consumer/producer created.

[merlimat]

The exception on acknowledge is an artifact of existing APIs. The current implementation is not throwing exceptions anymore (based on the fact that messages that are failed to be acknowledged will be replayed).

[codelipenghui]

In doAcknowledge method, if check state failed.

if (getState() != State.Ready && getState() != State.Connecting) {
            stats.incrementNumAcksFailed();
            PulsarClientException exception = new PulsarClientException("Consumer not ready. State: " + getState());
            if (AckType.Individual.equals(ackType)) {
                onAcknowledge(messageId, exception);
            } else if (AckType.Cumulative.equals(ackType)) {
                onAcknowledgeCumulative(messageId, exception);
            }
            return FutureUtil.failedFuture(exception);
        }

So i thought parameter of exception should be keep.

[merlimat]

This should actually only be called when the consumer has been closed.

[merlimat]

Any reason for why the message builder cannot be recycled anymore?

[codelipenghui]

ProducerIntercetpor want to get the message properties. The message properties is lazy init by call getProperties(). In getProperties method:

 public synchronized Map<String, String> getProperties() {
        if (this.properties == null) {
            if (msgMetadataBuilder.getPropertiesCount() > 0) {
                this.properties = Collections.unmodifiableMap(msgMetadataBuilder.getPropertiesList().stream()
                        .collect(Collectors.toMap(KeyValue::getKey, KeyValue::getValue)));
            } else {
                this.properties = Collections.emptyMap();
            }
        }
        return this.properties;
    }

So i change call msgBuilder.recycle() after call interceptor.

public void sendComplete(Exception e) {
                if (e != null) {
                    stats.incrementSendFailed();
                    onSendAcknowledgement(interceptorMessage, null, e);
                    future.completeExceptionally(e);
                } else {
                    onSendAcknowledgement(interceptorMessage, interceptorMessage.getMessageId(), null);
                    future.complete(interceptorMessage.getMessageId());
                    stats.incrementNumAcksReceived(System.nanoTime() - createdAt);
                }
                interceptorMessage.getDataBuffer().release();
                interceptorMessage.getMessageBuilder().recycle();
                while (nextCallback != null) {
                    SendCallback sendCallback = nextCallback;
                    MessageImpl<?> msg = nextMsg;
                    msg.getDataBuffer().retain();
                    if (e != null) {
                        stats.incrementSendFailed();
                        onSendAcknowledgement((Message<T>) msg, null, e);
                        sendCallback.getFuture().completeExceptionally(e);
                    } else {
                        onSendAcknowledgement((Message<T>) msg, msg.getMessageId(), null);
                        sendCallback.getFuture().complete(msg.getMessageId());
                        stats.incrementNumAcksReceived(System.nanoTime() - createdAt);
                    }
                    msg.getDataBuffer().release();
                    msg.getMessageBuilder().recycle();
                    nextMsg = nextCallback.getNextMessage();
                    nextCallback = nextCallback.getNextSendCallback();
                }
            }

[merlimat]

Can you add a comment on why we are retaining the buffer here and when it will be released?

[codelipenghui]

ok

[sijie]

@merlimat : created #2476 for tracking the tasks for interceptors. lets focus on producer/consumer interceptor for this PR.

One thing we should consider is to add ReaderInterceptor as well. Even though the implementation is completely based on consumer, from an API perspective they are completely different.

Created #2477

Another idea, which might be out of the scope of this PR is to have Client level interceptor, both to be automatically attached to any producer/consumer created by the client instance and to also intercept events like consumer/producer created.

Created #2478

[merlimat]

@merlimat : created #2476 for tracking the tasks for interceptors. lets focus on producer/consumer interceptor for this PR.

👍 Sounds good

[codelipenghui]

@merlimat I'm already address all your comments.

[merlimat]

👍

[sijie]

retest this please

[merlimat]

@codelipenghui There is one test failing : https://builds.apache.org/job/pulsar_precommit_java8/3387/testReport/junit/org.apache.pulsar.functions.instance/ContextImplTest/testPublishUsingDefaultSchema_on_testPublishUsingDefaultSchema_org_apache_pulsar_functions_instance_ContextImplTest_/

I believe it might be related to the test mocking the producer implementation and then find the interceptor variable as null.

[sijie]

run java8 tests

[sijie]

@codelipenghui I fixed ReplicatorTest and V1_ReplicatorTest. so the CI should be passing now. once the CI is passed, will merge it.

[sijie]

run java8 tests

[sijie]

@codelipenghui I just merged the interceptors change. Great work!

[codelipenghui]

@sijie Thanks.