Add producer purge method with optional blocking argument by peteryin21 · Pull Request #548 · confluentinc/confluent-kafka-python

Peter Yin (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.

Deleted user (ghost) · 2019-02-27T17:06:32Z

It looks like Peter Yin (@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

Ryan P (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

Ryan P (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.

Peter Yin (peteryin21) · 2019-03-06T17:16:14Z

[clabot:check]

Deleted user (ghost) · 2019-03-06T17:16:15Z

Confluent Inc. (@confluentinc) It looks like Peter Yin (@peteryin21) just signed our Contributor License Agreement. 👍

Always at your service,

clabot

Peter Yin (peteryin21) · 2019-03-06T17:43:14Z

Thanks for the review Ryan P (@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?

Ryan P (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.

Magnus Edenhill (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.

Peter Yin (peteryin21) · 2020-05-06T18:42:27Z

Magnus Edenhill (@edenhill) Ryan P (@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.

Magnus Edenhill (edenhill)

Good work!

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

Magnus Edenhill (edenhill) · 2020-05-08T07:06:34Z

confluent_kafka/admin/__init__.py

    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)

Magnus Edenhill (edenhill) · 2020-05-08T07:07:26Z

confluent_kafka/src/Producer.c

 }

+
+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!

Magnus Edenhill (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))

Magnus Edenhill (edenhill) · 2020-05-08T07:11:40Z

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

…ent to Producer_purge

Peter Yin (peteryin21) · 2020-06-18T01:49:51Z

Magnus Edenhill (@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

Peter Yin (peteryin21) · 2020-06-29T18:56:18Z

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

confluent_kafka/src/Producer.c

Peter Yin (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.

Magnus Edenhill (edenhill)

LGTM!

Magnus Edenhill (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.

Peter Yin (peteryin21) · 2020-07-13T20:37:37Z

Magnus Edenhill (@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?

Peter Yin (peteryin21) · 2020-07-30T18:41:58Z

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

Peter Yin (peteryin21) · 2020-10-15T17:03:19Z

Magnus Edenhill (@edenhill) Can we get this PR merged?

Magnus Edenhill (edenhill) · 2020-10-20T09:15:11Z

Thanks for this, Peter Yin (@peteryin21) !

Peter Yin (peteryin21) mentioned this pull request Feb 27, 2019

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

Closed

7 tasks

Ryan P (rnpridgeon) reviewed Mar 5, 2019

View reviewed changes

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

Ryan P (rnpridgeon) reviewed Mar 5, 2019

View reviewed changes

Ryan P (rnpridgeon) suggested changes Mar 5, 2019

View reviewed changes

Ryan P (rnpridgeon) closed this Apr 7, 2020

Magnus Edenhill (edenhill) reviewed May 8, 2020

View reviewed changes

Ryan P (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

Peter Yin (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

Peter Yin (peteryin21) requested review from Magnus Edenhill (edenhill) and Ryan P (rnpridgeon) 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

Magnus Edenhill (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

Peter Yin (peteryin21) requested a review from Magnus Edenhill (edenhill) July 7, 2020 21:16

Magnus Edenhill (edenhill) approved these changes Jul 7, 2020

View reviewed changes

Magnus Edenhill (edenhill) merged commit baee8a6 into confluentinc:master Oct 20, 2020

		DEFAULT_CONFIG = CONFIG_SOURCE_DEFAULT_CONFIG #:


		class PurgeStrategy(Enum):

Conversation

Peter Yin (peteryin21) commented Feb 27, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Background

Description

Uh oh!

Deleted user (ghost) commented Feb 27, 2019

Uh oh!

Uh oh!

Ryan P (rnpridgeon) Mar 5, 2019

Choose a reason for hiding this comment

Uh oh!

Ryan P (rnpridgeon) left a comment

Choose a reason for hiding this comment

Uh oh!

Peter Yin (peteryin21) commented Mar 6, 2019

Uh oh!

Deleted user (ghost) commented Mar 6, 2019

Uh oh!

Peter Yin (peteryin21) commented Mar 6, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Ryan P (rnpridgeon) commented Apr 7, 2020

Uh oh!

Magnus Edenhill (edenhill) commented Apr 8, 2020

Uh oh!

Peter Yin (peteryin21) commented May 6, 2020

Uh oh!

Magnus Edenhill (edenhill) left a comment

Choose a reason for hiding this comment

Uh oh!

Magnus Edenhill (edenhill) May 8, 2020

Choose a reason for hiding this comment

Uh oh!

Magnus Edenhill (edenhill) May 8, 2020

Choose a reason for hiding this comment

Uh oh!

Magnus Edenhill (edenhill) May 8, 2020

Choose a reason for hiding this comment

Uh oh!

Magnus Edenhill (edenhill) commented May 8, 2020

Uh oh!

Peter Yin (peteryin21) commented Jun 18, 2020

Uh oh!

Peter Yin (peteryin21) commented Jun 29, 2020

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Peter Yin (peteryin21) commented Jul 7, 2020

Uh oh!

Magnus Edenhill (edenhill) left a comment

Choose a reason for hiding this comment

Uh oh!

Magnus Edenhill (edenhill) commented Jul 7, 2020

Uh oh!

Peter Yin (peteryin21) commented Jul 13, 2020

Uh oh!

Peter Yin (peteryin21) commented Jul 30, 2020

Uh oh!

Peter Yin (peteryin21) commented Oct 15, 2020

Uh oh!

Magnus Edenhill (edenhill) commented Oct 20, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Peter Yin (peteryin21) commented Feb 27, 2019 •

edited

Loading

Peter Yin (peteryin21) commented Mar 6, 2019 •

edited

Loading