Add producer purge method with optional blocking argument #548

peteryin21 · 2019-02-27T17:06:30Z

Background

Currently, if a producer times out (i.e when a broker goes down), messages are added to an internal queue to be published. The number of remaining messages is returned via the flush() command.
In use cases where duplicate messages might be highly undesirable (e.g sending task requests to a factory), it would be useful to purge failed messages from the internal queue so they are not unintentionally sent the next time flush() is called with the brokers back online.

Description

Luckily, librdkafka exposes a purge api that purges messages in internal queues. This PR adds a binding for this functionality to this library so we can access it via the python Producer class. There are three strategies of purging that are defined by the PurgeStrategy Enum that can be passed in as first positional argument.

PURGE_INTERNAL_QUEUES: Purge messages from internal queues.
PURGE_INFLIGHT: Purge messages in flight to or from the broker.
PURGE_ALL: Purge messages from internal queues and those in flight.
By default, the method will block, but if the optional argument blocking is set to False, we will not wait on the background thread queue.

In addition, the user will need to call poll() or flush() after purge() to serve the delivery report callbacks of the purged messages. I conservatively decided to leave it out, but am open to discussion for how to best make this intuitive to the user.

ghost · 2019-02-27T17:06:32Z

It looks like @peteryin21 hasn't signed our Contributor License Agreement, yet.

The purpose of a CLA is to ensure that the guardian of a project's outputs has the necessary ownership or grants of rights over all contributions to allow them to distribute under the chosen licence.
Wikipedia

You can read and sign our full Contributor License Agreement here.

Once you've signed reply with [clabot:check] to prove it.

Appreciation of efforts,

clabot

confluent_kafka/src/Producer.c

rnpridgeon · 2019-03-05T11:53:58Z

confluent_kafka/src/Producer.c

+	if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|p", kws, &blocking))
+			return NULL;
+	if (blocking==1)
+		rd_kafka_purge(rk, RD_KAFKA_PURGE_F_QUEUE)


Make sure to handle any errors which may be raised from the execution of rd_kafka_purge

https://github.com/edenhill/librdkafka/blob/master/src/rdkafka.h#L3478-L3482

rnpridgeon

Overall it looks like you are the right track. We need to add error handling and we should expose the full purge implementation including the ability to purge in flight request.

peteryin21 · 2019-03-06T17:16:14Z

[clabot:check]

ghost · 2019-03-06T17:16:15Z

@confluentinc It looks like @peteryin21 just signed our Contributor License Agreement. 👍

Always at your service,

clabot

peteryin21 · 2019-03-06T17:43:14Z

Thanks for the review @rnpridgeon! I've made the requested changes and fixed the build errors. Let me know how it looks.

Also, I think it would be worth adding some tests, but wanted to get your thoughts. I was thinking of adding a def test_purge to test_Producer.py and adding purge() usage to this test in integration_test.py. What do you think?

rnpridgeon · 2020-04-07T20:55:57Z

Sorry I didn't get back to this in time. At any rate I believe the Transactional Producer API is better suited for this. I'll close this for now but if we deem the Transactional API to not be sufficient we can revisit.

edenhill · 2020-04-08T07:14:34Z

purge() is useful for other purposes, for instance when terminating and the application wants to know the exact status of each message delivery without having to wait the full message.timeout.ms on flush.

E.g., something like this:

# terminate

p.flush(5)  # wait for messages to be delivered, but much shorter than message.timeout.ms

# purge any remaining messages from queues/inflight
p.purge(in_queue=True, in_flight=True)

p.flush(0)  # get delivery reports for the purged messages

sys.exit(0)

I think we should reconsider adding this functionality since Peter has put the effort in to implement it.

peteryin21 · 2020-05-06T18:42:27Z

@edenhill @rnpridgeon It's been a while! I will look more into the transactional producer api. In the meantime, if purge() would still be a useful method to expose, then I'm happy to try and get this across the finish line. Let me know what the remaining steps are before we can merge this.

edenhill

Good work!

I think we can simplify the implementation and the API by not exposing the enums/flags and simply have bool kwargs.

edenhill · 2020-05-08T07:06:34Z

confluent_kafka/admin/__init__.py

@@ -35,6 +38,15 @@ class ConfigSource(Enum):
    DEFAULT_CONFIG = CONFIG_SOURCE_DEFAULT_CONFIG  #:


+class PurgeStrategy(Enum):


While this technically maps nicely to librdkafka, I think from a user-experience it might be better to simply go with optional boolean kwargs: purge(in_queue=True, in_flight=True, blocking=True) (those are the defaults)

edenhill · 2020-05-08T07:07:26Z

confluent_kafka/src/Producer.c

@@ -398,6 +398,34 @@ static PyObject *Producer_flush (Handle *self, PyObject *args,
        return cfl_PyInt_FromInt(qlen);
 }

+
+static void *Producer_purge (Handle *self, PyObject *args,


I know the source is messed up tab/space-wise, but please try to use 8-space indents in new code, no tabs.
Thank you!

edenhill · 2020-05-08T07:10:14Z

confluent_kafka/src/Producer.c

+	if (err == RD_KAFKA_RESP_ERR_NO_ERROR) {
+		Py_RETURN_NONE;
+	}
+	if (err == RD_KAFKA_RESP_ERR__INVALID_ARG) {


I don't think the specific error code handling is needed here, it is very unlikely to happen, so to keep it simple simply do cfl_PyErr_format(err, "%s", rd_kafka_err2str(err))

edenhill · 2020-05-08T07:11:40Z

It also needs a unit-test (tests/test_Producer.py)

…ent to Producer_purge

peteryin21 · 2020-06-18T01:49:51Z

@edenhill Thanks for the comments. I've addressed them and also added a new test for purge in test_Producer.py. It seems to work! The things I don't think I can test functionally are the blocking and in_flight flags, as well as the error, since these will probably require integration testing. Let me know what you think

peteryin21 · 2020-06-29T18:56:18Z

@edenhill Ping in case this fell off your radar :)

confluent_kafka/src/Producer.c

peteryin21 · 2020-07-07T21:17:56Z

Looks like there are some issues with prepare-osx.sh that are also being experienced in recent PRs #909. I can rebase this branch on top of the fix once that's figured out.

edenhill

LGTM!

edenhill · 2020-07-07T21:23:56Z

Let's get #909 merged first, then we'll restart your CI jobs to make sure they pass before merging.

peteryin21 · 2020-07-13T20:37:37Z

@edenhill It looks like #909 is merged -- would you be able to restart the CI jobs and then merge?

I'm also curious what the release schedule looks like, and what the timeline for releasing this change might look like?

peteryin21 · 2020-07-30T18:41:58Z

Hey @edenhill, would it be possible to restart the CI jobs and then merge if it looks good?

peteryin21 · 2020-10-15T17:03:19Z

@edenhill Can we get this PR merged?

edenhill · 2020-10-20T09:15:11Z

Thanks for this, @peteryin21 !

peteryin21 mentioned this pull request Feb 27, 2019

Can I clear the Producer output queue when flush() times out? #538

Closed

7 tasks

rnpridgeon reviewed Mar 5, 2019

View reviewed changes

confluent_kafka/src/Producer.c Outdated Show resolved Hide resolved

rnpridgeon reviewed Mar 5, 2019

View reviewed changes

rnpridgeon suggested changes Mar 5, 2019

View reviewed changes

rnpridgeon closed this Apr 7, 2020

edenhill reviewed May 8, 2020

View reviewed changes

rnpridgeon reopened this May 14, 2020

Peter Yin added 4 commits June 16, 2020 14:37

add producer purge method with optional blocking argument

8dbc229

Add PurgeStrategy enum exposing purge flags and add as required argum…

c8a7bb2

…ent to Producer_purge

Add error handling for Producer_purge

e0be387

misc fixes to errors and warnings when building

8dcad18

peteryin21 force-pushed the add_purge_to_producer branch from 0d008a1 to 8dcad18 Compare June 16, 2020 21:37

Peter Yin added 5 commits June 17, 2020 15:00

address comments, remove enum and make each option a bool

13e3636

This bug took way too long to figure out

71caead

fix another typo in purge kw list

07999b1

add test for purge in test_Producer.py

8b585b1

minor doc change

4880862

Peter Yin added 2 commits June 17, 2020 18:52

move declaration of purge strategy variable to top

1cce53c

Really python2? You don't understand underscores in numbers? Psh

a1d4ec6

peteryin21 requested review from rnpridgeon and edenhill June 18, 2020 16:50

Peter Yin added 2 commits June 18, 2020 11:53

python2 doesn't have the nonlocal keyword either

55eaa4f

python2 doesn't have 'p' format char

273ec69

edenhill suggested changes Jul 7, 2020

View reviewed changes

confluent_kafka/src/Producer.c Outdated Show resolved Hide resolved

confluent_kafka/src/Producer.c Outdated Show resolved Hide resolved

confluent_kafka/src/Producer.c Outdated Show resolved Hide resolved

confluent_kafka/src/Producer.c Outdated Show resolved Hide resolved

address comments from review

b1ffd0e

peteryin21 requested a review from edenhill July 7, 2020 21:16

edenhill approved these changes Jul 7, 2020

View reviewed changes

edenhill merged commit baee8a6 into confluentinc:master Oct 20, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add producer purge method with optional blocking argument #548

Add producer purge method with optional blocking argument #548

peteryin21 commented Feb 27, 2019 •

edited

Loading

ghost commented Feb 27, 2019

rnpridgeon Mar 5, 2019

rnpridgeon left a comment

peteryin21 commented Mar 6, 2019

ghost commented Mar 6, 2019

peteryin21 commented Mar 6, 2019 •

edited

Loading

rnpridgeon commented Apr 7, 2020

edenhill commented Apr 8, 2020

peteryin21 commented May 6, 2020

edenhill left a comment

edenhill May 8, 2020

edenhill May 8, 2020

edenhill May 8, 2020

edenhill commented May 8, 2020

peteryin21 commented Jun 18, 2020

peteryin21 commented Jun 29, 2020

peteryin21 commented Jul 7, 2020

edenhill left a comment

edenhill commented Jul 7, 2020

peteryin21 commented Jul 13, 2020

peteryin21 commented Jul 30, 2020

peteryin21 commented Oct 15, 2020

edenhill commented Oct 20, 2020

		@@ -35,6 +38,15 @@ class ConfigSource(Enum):
		DEFAULT_CONFIG = CONFIG_SOURCE_DEFAULT_CONFIG #:


		class PurgeStrategy(Enum):

Add producer purge method with optional blocking argument #548

Add producer purge method with optional blocking argument #548

Conversation

peteryin21 commented Feb 27, 2019 • edited Loading

Background

Description

ghost commented Feb 27, 2019

rnpridgeon Mar 5, 2019

Choose a reason for hiding this comment

rnpridgeon left a comment

Choose a reason for hiding this comment

peteryin21 commented Mar 6, 2019

ghost commented Mar 6, 2019

peteryin21 commented Mar 6, 2019 • edited Loading

rnpridgeon commented Apr 7, 2020

edenhill commented Apr 8, 2020

peteryin21 commented May 6, 2020

edenhill left a comment

Choose a reason for hiding this comment

edenhill May 8, 2020

Choose a reason for hiding this comment

edenhill May 8, 2020

Choose a reason for hiding this comment

edenhill May 8, 2020

Choose a reason for hiding this comment

edenhill commented May 8, 2020

peteryin21 commented Jun 18, 2020

peteryin21 commented Jun 29, 2020

peteryin21 commented Jul 7, 2020

edenhill left a comment

Choose a reason for hiding this comment

edenhill commented Jul 7, 2020

peteryin21 commented Jul 13, 2020

peteryin21 commented Jul 30, 2020

peteryin21 commented Oct 15, 2020

edenhill commented Oct 20, 2020

peteryin21 commented Feb 27, 2019 •

edited

Loading

peteryin21 commented Mar 6, 2019 •

edited

Loading