publish time non-monotonic on same partition

[fernandezpablo85]

Describe the bug
Publish time is not monotonic on the same partition.

To Reproduce
Steps to reproduce the behavior:

1. Publish events with a single producer to a topic with many partitions, with any partition key.
2. Run a consumer on key_shared mode, for example:

client = pulsar.Client(pulsar_url)
    consumer = client.subscribe(
        topic=pulsar_topic,
        subscription_name="any_name_you_want",
        consumer_name="any_name_you_want",
        consumer_type=ConsumerType.KeyShared,
        initial_position=InitialPosition.Earliest,
    )

    for i in itertools.count():
        msg = consumer.receive()
        ts = msg.publish_timestamp()
        p_id = msg.message_id().partition()
        lpt = latest_partition_timestamp[p_id]
        assert ts >= lpt, f"latest timestamp of partition {p_id} was {lpt} and new event has timestamp {ts}"
        latest_partition_timestamp[p_id] = ts
        consumer.acknowledge(msg)

3. See assertion error

Expected behavior
No error should be raised since publish_timestamp on the same partition must be monotonically increasing.

Desktop (please complete the following information):

• OS: macOS 10.15.2

Additional context

• producer: https://github.com/apache/pulsar-client-go
• consumer: https://pypi.org/project/pulsar-client/
• partitions: 3

[fernandezpablo85]

Note that this also fails when consuming from a single partition, by changing the topic to $TOPIC_NAME-partition-0 for example

[jiazhai]

@fangpengcheng95 Thanks for the report. Seems This is a issue producer in pulsar-client-go

[fernandezpablo85]

thanks @jiazhai , I'd be glad to contribute a patch if you give me a few pointers