feat: Implement initial pulsar support

[jaysonsantos]

trafficstars

Description

This adds an initial support for pulsar

Type of change

• [x] New feature (non-breaking change which adds functionality)
• [x] This change requires a documentation update

How Has This Been Tested?

Please describe the tests that you ran to verify your changes. Provide instructions so we can reproduce. Please also list any relevant details for your test configuration

Initial integration tests were run with the following:

import threading

import pulsar
from pulsar import ConsumerType, InitialPosition


def main():
    client = pulsar.Client("pulsar://localhost:6650")
    print(client)
    producer = client.create_producer("sample")
    
    producer.send(b"hello world")

    consumer = client.subscribe("sample", "consumer-1", consumer_type=ConsumerType.KeyShared, initial_position=InitialPosition.Earliest)
    print(consumer.receive(1_000))

    done = threading.Event()

    def consume(consumer, message):
        print('Consumer', consumer, 'Message', message)
        done.set()
    consumer2 = client.subscribe("sample", "consumer-2", message_listener=consume, consumer_type=ConsumerType.KeyShared, initial_position=InitialPosition.Earliest)
    consumer2.resume_message_listener()

    done.wait()
    consumer2.close()
    consumer.close()

    producer.close()
    client.close()

Does This PR Require a Core Repo Change?

• [ ] Yes. - Link to PR:
• [x] No.

Checklist:

See contributing.md for styleguide, changelog guidelines, and more.

• [x] Followed the style guidelines of this project
• [ ] Changelogs have been updated
• [x] Unit tests have been added
• [ ] Documentation has been updated