Walk around the Pulsar deduplication check

[gaoran10]

Motivation

Currently, the AoP couldn't work with the Pulsar deduplication feature, and it seems the AMQP protocol doesn't support the message deduplication feature. So we could walk around the Pulsar deduplication check when publishing messages in AoP.

Modifications

Add a new class AmqpEntryWriter to process messages publishing, and use the method ManagedLedger#asyncAddEntry instead of PersistentTopic#publishMessage to write messages.

Verifying this change

Add a new test to verify the AoP could work well when enabling message deduplication in Pulsar.

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): (no)
• The public API: (no)
• The schema: (no)
• The default values of configurations: (no)
• The wire protocol: (no)
• The rest endpoints: (no)
• The admin cli options: (no)
• Anything that affects deployment: (no)

Documentation

Check the box below.

Need to update docs?

• doc-required

  (If you need help on updating docs, create a doc issue)

• no-need-doc

  (Please explain why)

• doc

  (If this PR contains doc changes)