{"push_id":3174571922,"size":1,"distinct_size":1,"ref":"refs/heads/feature/delayed_consumers","head":"ea958592905d864efd693e65429797b528a39826","before":"bdce3444b62e68a58488f002955c2a67ceeea4c7","commits":[{"sha":"ea958592905d864efd693e65429797b528a39826","author":{"name":"Ezequiel Lovelle","email":"a787c4be85dd0f0c610e28a432436de10dfe898a@gmail.com"},"message":"Deferred messages for consumers\n\nThis feature offers the capability to configure consumer subscription with an\narbitrary receiver delay.\n\n[pulsar-broker]\n  - Add field for DelayQueue to store next positions pending of delivery.\n\n  - Add field for ScheduledFuture to schedule the head of DelayQueue in order to\n    start the process of delivering expired entries.\n\n  - Fix on sendMessages method usage of received list of entries when such\n    entries are being filtered by updatePermitsAndPendingAcks method.\n    Received list of entries might be filtered by updatePermitsAndPendingAcks\n    method and afterwards entries are processed by execute() with lambda which\n    is probably that it's executing thread will be other than its calling thread.\n\n    Therefore entries are now wrapped with CopyOnWriteArrayList in order\n    to prevent ConcurrentModificationException by lambda on execute(), another\n    approach could be to copy the entire list for lambda or to use a synchronized\n    list, but this would result in a performance penalty even when entries are\n    not being filtered, CopyOnWriteArrayList prevent this from happening.\n\n    Another step further trying to fix this might be using Streams and applying\n    transformations to inner list. This path was not taken because would\n    require major changes.\n\n  - Add processDelayEntries() method to process all elements added in DelayQueue\n    which are ready to be delivered, at any given time just one task should be\n    schedule using this method.\n\n  - Add readPublishFrom() method to get the parameter of publish time from\n    metadata  of a message without changing its reference offset, this method\n    will only be used if the consumer has enabled the receiver delay parameter\n    on the subscription.\n\n  - Add inner private class DelayPositionInfo to represent each position to be\n    schedule in DelayQueue.\n\n  - Add method to clean-up previous mentioned fields related to deferred\n    messages.\n\nDelayQueue from java.util.concurrent is used in order to store each messages\nposition next to be expired, the advantage of using this queue is that at any\ngiven time one and only one task is scheduled per consumer avoiding to schedule\nan unbounded number of tasks.\n\n[pulsar-client]\n  - Add receiverDelay() method at subscription level to configure this parameter.\n\n[pulsar-common]\n  - Set receiver delay whether it was configured by user on subscription.\n\n  - Add optional receiver delay parameter to protobuf pulsar schema,\n    code generated by generate_protobuf_docker.sh script.","distinct":true,"url":"https://api.github.com/repos/lovelle/pulsar/commits/ea958592905d864efd693e65429797b528a39826"}]}
{"actor":{"display_login":"lovelle"}}