Support set publish time on broker side

[lovelle]

Motivation

Since #3155 there is a real need not to depend on client side clock for publish time field on message metadata.

This change sets a default value for publish time field at client side in order to be able to be identified by broker easily and set newly calculated time on it. If time was normally set by the client on publish time, this update will be just ignored by broker.

Modifications

[pulsar-broker]

• Read message metadata for produced messages in order to know whether the
  message was set onto default value and should be updated with time
  calculated by broker.
• Add method to return previous metadata with publish time set by broker.

[pulsar-client]

• ProducerImpl client side set a default fixed size value for publish time.

[pulsar-common]

• Add modifyMessageMetadata method on Commands in order to update new modified
  fields on payload.

The default constant value used to set the undefined number of milliseconds since epoch is going to be 13 decimal digits, the method currentTimeMillis will always return 13 decimal digits assuming the broker has the current time set correctly.

So we are safe with the assumption that the return value is 13 decimal digits for more than the next two centuries.

Result

Any client which set publish time with default value will be updated by broker with current time.

TODO

Once this change is accepted:

• Add proper tests
• Add default publish time on the rest of clients.

[merlimat]

Just leaving a couple of comments and considerations:

1. The reason we're currently relying on the client specified publish time is that is "cheaper". In broker we don't have to re-serialize the protobuf metadata, but rather we pass the same unmodified buffer to BK client. In any case, I'd like to have the option to turn on/off this feature based on needs.
2. Since there can be clock skew across brokers as well, we would also need to make sure that the timespace is always guaranteed to be monotonically increasing, in particular after a broker failover. (eg: after failover set publish-timestamp to be >= to the timestamp of last written message in the topic
3. Right now, there is no option to set the publish timestamp (as opposed to the event time which is optional). Rather the publish timestamp is always automatically set in the client library. If we have the broker-set publish timestamp, we should always override what was passed by client. Another option, could be to leave the current publish-timestamp (or rename it, since that is fine in protobuf) and use a different field for the broker-set timestamp. When doing some operations (eg: TTL, delayed delivery, etc), we can use the broker-set timestamp if present or fallback to client-set otherwise.
4. Continuing from the above point, we also need to consider this timestamp when doing async-replication. When the message is republished, it will need to have a new timestamp, to make sure the monotonicity is maintained.

[merlimat]

Instead of constants, we should just use the hasPublishTime() method that will report if the field was set in the serizalized protobuf binary.

[lovelle]

Sorry, I didn't explained why I did it that way.

The reason I used a constant of 13 digits size is because the method modifyMessageMetadata() should be able to just update this value without realloc the entire buffer due to being with different size, the problem with hasPublishTime() is that if it is false then publish time on metadata is literally '0' and the new epoch will be of 13 digits length.

The alternative approach I can think to get rid of the constant definition is to set publish time field on protobuf with the default value of 13 digits and testing if publish time has such default comparing with a new metadata created.
Maybe proto3 solved this? not really sure.

[lovelle]

Thanks for your comments, I answer you each point:

1. I agree that is not acceptable to deserialize and serialize the entire payload just to change publish time (or any field). But, AFAIK, on the added method modifyMessageMetadata() the entire payload is just read until message metadata (which has a maximum possible size of 1Kb) and write the uint64 to the buffer (new epoch) and update the new checksum if present, I don't think this is a considerable overhead, please correct me if I'm wrong.
   I agree this behavior should be allowed to be turned on/off by client.

2. Yep, nice catch up. Therefore, only the greater epoch would be the one present on metadata, isn't it?

3. I think the answer to this might be 'it depends'. It depends in the way you think about the 'delayed messages' feature

   I think adding a new field is a good solution, the only issue I can see here is that the consumer is related with the decision made by the producer of setting this new field, and the main goal of Deferred messages for consumers #3155 is to make delay on consumer messages regardless of the producer.
   Anyway, we can add a new optional field to set this 'publish time' set by default by the client, maybe with the ability to be modified by the user if they want? Another downside of this would be the producer must update pulsar client too and the consumer should be aware that the producer has been updated with new version.

4. Yep, but depends on the meaning given to this specific field, either the current publish time or a new one, it depends if it means the epoch since the message has been created or the epoch since the user has created the message, for the former one we need to do what you said (do we have something that identifies the message as a published by replication?), but if it is the latter case I think we don't need to do anything more.

[sijie]

@merlimat can you check @lovelle 's last comment?

[sijie]

@merlimat can you check @lovelle 's last comment?

[lovelle]

@sijie @merlimat I think this pr should be reviewed very carefully. IMHO #3155 could exist without this change, maybe with a warning to the user about publish time being client generated.
We discussed about delayed messages on dev mailing list and I think it would be great if we (especially all of you the leaders) could get general consensus if we want #3155 to be on pulsar or not.

Thanks!

[sijie]

@lovelle

sorry for forgetting to follow up with your changes. IMO supporting setting publish time on broker is a great feature to have and would be very useful for many other use cases which require monotonically increasing "time". so I think the PR should not be considered just for #3155, it has a broader set of use cases to support. We should discuss this feature independent of #3155.

I think general consensus of delayed message was using functions to achieve that. but we didn't have a consensus of #3155 (IMO it is a "delayed" subscription feature). I will ping the thread and let's come up with a consensus. IMO "delayed" subscription can be achieved with the proposal in #3155

/cc @merlimat

[sijie]

@lovelle Just sent an email to that email thread to try to conclude the solution and get the community to achieve an agreement

[lovelle]

@sijie Excellent! fully agree with you! Thank you very much for the follow up 👍