KAFKA-3888 Use background thread to process consumer heartbeats #1266

dpkp · 2017-10-18T00:18:48Z

This PR implements KAFKA-3888 (including related KAFKA-4431). It adds significant complexity and relies on managing shared mutable state, which I really don't feel great about. Nonetheless, I wanted to put this up so that folks can see where it's at and possibly test out independently.

One remaining issue: address default configuration changes when broker does not support rebalance timeout (only session timeout).

ref #948 / includes #1258
See also KAFKA-4160

jeffwidman

Thanks for all the hard work on this.

Left a few comments. I need to re-read the KIP to understand the config parameters a little better...

jeffwidman · 2017-10-19T03:30:12Z

kafka/client_async.py

@@ -512,46 +527,40 @@ def poll(self, timeout_ms=None, future=None, delayed_tasks=True):
        Returns:
            list: responses received (can be empty)
        """
-        if timeout_ms is None:
+        if future is not None:
+            timeout_ms = 100


I don't understand what this does. What are the side effects of hardcoding this value here?

@dpkp This looks more like a hack actually. Skipped it in the parent PR, but yea...

Yes, deserves a comment. The issue is that we now have to account for the future being resolved in a different thread.

Prior to this change we would block on network IO and then check the future's completion after processing the network IO and any corresponding futures. But now that we have separated future handling from network IO, and we have (intentionally) moved the future handling into a separate section that is not locked, we can have a situation where one thread is called with a simple timeout and another is called with a future (block until resolved).

The first thread acquires the lock and receives the response that will resolve the future in its _poll() call. The response is put on the internal pending_completions queue and the lock is released. Now the second thread takes the lock and checks whether the future is resolved. It is not, so it calls _poll with a full request timeout, which is now 5 minutes. The first thread continues, without the lock, and begins processing pending_completions. The future is resolved with the queued response, and the first thread finishes. But the second thread is now waiting for network IO that it thinks is necessary to complete its future. Except that the future is already done and so there is no network IO coming. It will simply wait for 5 minutes and then timeout before rechecking the future and then finally returning.

My solution here is to simply reduce the network IO timeout when called with a future to reduce the unneeded block time when this occurs. Another solution might be to try to register any future that is "blocking" and call _wake_up() on when such a future is resolved. I decided against that approach because I think it is too complex and it is likely to add overhead to every response, not just ones for which some thread is blocking on a future.

I am glad I asked. I never would have realized that. Add this explanation as a code comment would be much appreciated.

jeffwidman · 2017-10-19T03:45:38Z

kafka/client_async.py

@@ -721,6 +739,7 @@ def _maybe_refresh_metadata(self):
        Returns:
            int: milliseconds until next refresh
        """
+        # This should be locked when running multi-threaded


This comment is slightly ambiguous.

What does This refer to? I assume self.cluster.ttl(), but possibly you meant a broader scope than that.

And who is responsible for enforcing the locking? The caller of set_topics() or the implementation of self.cluster.ttl()?

Comment relates to entire function, which is currently only called inside a locked section. Will update.

jeffwidman · 2017-10-19T03:51:48Z

kafka/client_async.py

-            task (callable): task to be unscheduled
-        """
-        self._delayed_tasks.remove(task)
-


Why does adding background heartbeat enable task scheduling to be removed? Aren't there still have tasks that need to be scheduled for execution at some point in the future, such as a metadata refresh? Or, in the case of metadata refresh, is that already checked every time we call poll() and hence doesn't need to be scheduled?

The other "tasks" are metadata refresh and offset commits. These are now handled inline during client.poll and coordinator.poll, respectively.

jeffwidman · 2017-10-19T03:55:01Z

kafka/conn.py

@@ -704,7 +716,7 @@ def can_send_more(self):
    def recv(self):
        """Non-blocking network receive.

-        Return response if available
+        Return list of (response, future)


adding the word "tuples" improves readability: Return list of (response, future) tuples

jeffwidman · 2017-10-19T04:07:19Z

kafka/coordinator/base.py

+    class's monitor. Generally this means acquiring the lock before reading or
+    writing the state of the group (e.g. generation, member_id) and holding the
+    lock when sending a request that affects the state of the group
+    (e.g. JoinGroup, LeaveGroup).


Fantastic job with the docs, as always.

jeffwidman · 2017-10-19T05:34:24Z

kafka/coordinator/consumer.py

+            # metadata is fresh, any metadata update that changes the topic
+            # subscriptions and arrives with a rebalance in progress will
+            # essentially be ignored. See KAFKA-3949 for the complete
+            # description of the problem.


Does this PR also fix #1241?

No, that will need a separate PR

jeffwidman · 2017-10-19T05:35:30Z

kafka/coordinator/consumer.py

+        """
+        Return the time to the next needed invocation of {@link #poll(long)}.
+        @param now current time in milliseconds
+        @return the maximum time in milliseconds the caller should wait before the next invocation of poll()


docstring will need to be updated from Java to python

jeffwidman · 2017-10-19T05:53:40Z

kafka/coordinator/consumer.py

+                callback, offsets, exception = self.completed_offset_commits.popleft()
+                callback(offsets, exception)
+        except IndexError:
+            pass


Is the try/except really needed? Why not while self.completed_offset_commits:?

yes, that seems cleaner

jeffwidman · 2017-10-19T06:18:41Z

kafka/errors.py

+            typically implies that the poll loop is spending too much
+            time message processing. You can address this either by
+            increasing the session timeout or by reducing the maximum
+            size of batches returned in poll() with max.poll.records.


This error message will need to be tweaked slightly to handle brokers that do/don't support heartbeating so that end users know what knobs to adjust. This will in turn depend on how max_poll_interval_ms is interpreted for brokers that don't support heartbeating... whether it is ignored or used in conjunction with session_timeout_ms somehow...

Also the param names need to be switched from Java to python.

Good call on updating the text. But the CommitFailed error is only raised when using group coordinated groups. It will not be seen when using zookeeper offsets from 0.8.2. That said, it can be seen by users that do not have max_poll_interval_ms support (0.9 <= broker < 0.10.1), and so it may need some tweaking when we decide how to manage max_poll_interval_ms in that context.

jeffwidman · 2017-10-19T10:16:28Z

test/test_coordinator.py

-    #(OffsetFetchResponse[0]([('foobar', [(0, 123, b'', 7), (1, 234, b'', 7)])]),
-    # Errors.RequestTimedOutError, True, False),
-    #(OffsetFetchResponse[0]([('foobar', [(0, 123, b'', 27), (1, 234, b'', 27)])]),
-    # Errors.RebalanceInProgressError, False, True),


What was the initial purpose of these? Why were they commented out?

I'm not sure and so I deleted them.

tvoinarovskyi · 2017-10-21T14:36:14Z

Will try to do a review tomorrow. Sorry for the holdup.

tvoinarovskyi

I did go through, but it's a bit too big to reason on the full scale. We should give it some time to just sit there after the merge, something is bound to pop up.
Any ideas how to actually test the feature? I didn't find any integration test on the feature.

tvoinarovskyi · 2017-10-22T03:50:32Z

kafka/coordinator/base.py

+
+            if self._heartbeat_thread is None:
+                log.debug('Starting new heartbeat thread')
+                self._heartbeat_thread = HeartbeatThread(weakref.proxy(self))


Why do you use the weakref.proxy here? Do we even need it as a weak reference?

There is a circular reference between coordinator <-> heartbeat_thread. The weakref here means that the existence of the heartbeat_thread will not prevent the coordinator from being garbage collected.

tvoinarovskyi · 2017-10-22T04:04:18Z

kafka/coordinator/base.py

+                self.disable()
+                return
+
+            # When consumer.wakeup() is implemented, we need to


Add XXX or FIXME here, so we don't lose it.

tvoinarovskyi · 2017-10-22T04:07:59Z

kafka/coordinator/base.py

+                self.coordinator.heartbeat.sent_heartbeat()
+                future = self.coordinator._send_heartbeat_request()
+                future.add_callback(self._handle_heartbeat_success)
+                future.add_errback(self._handle_heartbeat_failure)


Yea, you can get that from reading Java code. It never shrinks, parts are only added on top. Probably because a lot of people are working on it, but really is hard to follow sometimes.

tvoinarovskyi · 2017-10-22T04:24:43Z

Great job on this!

dpkp · 2017-10-22T16:33:32Z

So this is a very large PR. I think it would be possible to break it into a few smaller parts (locks, MemberState/Generation, coordinator.poll, heartbeat_thread). The benefit of keeping it together is that it is easier to cross-reference with the java PR that implements the same.

Re testing, I think the existing test suite covers a fair amount of the raw functionality. But we should add some specific tests for heartbeating while not polling. Separately, I am very interested in refactoring to a sans-io core that can be more easily unit tested without requiring fixtures or mocking. But I'll leave that for another day (I've been working on a simple state-machine representation of the group coordinator in a separate branch).

tvoinarovskyi · 2017-10-23T08:08:11Z

@dpkp - is the configuration change backward compatible? If I have a consumer configured with a custom 'session_timeout_ms' will it not break after update? As far as I see it will require to add 'max_poll_ms'...

tvoinarovskyi · 2017-10-23T08:10:12Z

As for the merge, I think it's ok as a big PR. It is not something that has meaning in merging separately.

dpkp · 2017-10-23T16:21:09Z

@tvoinarovskyi - I put the compatibility logic into KafkaConsumer and left the Coordinator classes to simply enforce constraints. KafkaConsumer will check the api_version after KafkaClient bootstrap, and if < 0.10.1 (max_poll_interval_ms not supported in JoinGroupRequest) then it will set max_poll_interval_ms = session_timeout_ms . In addition, if the user provides max_poll_interval_ms but not session_timeout_ms, we use the same value for session_timeout_ms. If the user provides neither, we use 30000/30000 defaults instead of the 10000/300000 used when max_poll_interval_ms is supported.

jeffwidman · 2017-11-07T23:12:46Z

What are next steps here? @dpkp are there still features/tests you're planning to add when you get time or is this ready to merge to master?

dpkp · 2017-11-07T23:55:17Z

I had hoped to deploy this build to production in some isolated services and verify that it performs as expected. I haven't had a chance to do that yet. Other than that I think it's ready to land. (sans merge conflict)

jeffwidman · 2017-11-08T00:33:43Z

Sounds good. I've been hoping to do a few impromptu tests myself, particularly in a couple of services we have that use gevent, but haven't had a chance either...

tvoinarovskyi · 2017-12-17T12:27:37Z

@dpkp Hey there, this PR will only get harder to rebase later. Is there any reason besides testing to not merge it? If it's merged we can at least find some feedback from others if it works as expected.

dpkp · 2017-12-17T14:28:31Z

I think only because I have been travelling and offline quite a bit, so would not be able to provide quick support if bugs we're found. I am ok with merging.

…

On Dec 17, 2017 7:27 AM, "Taras Voinarovskyi" ***@***.***> wrote: @dpkp <https://github.com/dpkp> Hey there, this PR will only get harder to rebase later. Is there any reason besides testing to not merge it? If it's merged we can at least find some feedback from others if it works as expected. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1266 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAzetGYlnoKGuWA1vMTeUupY1EGiwFKSks5tBQi6gaJpZM4P8_g1> .

dpkp · 2017-12-20T18:28:59Z

I rebased + fixed conflicts. I'm planning to merge after travis tests pass

…ns in coordinator.poll()

… needs next poll

… for old brokers

jeffwidman · 2017-12-21T22:59:38Z

Thanks again for all the hard work on this massive PR.

everpcpc · 2017-12-28T09:50:10Z

thanks for the hard work 👍

sanfilippopablo · 2018-04-12T12:58:14Z

Is this already out in the wild in some kafka-python version?
From what I understand, this will allow me to have long processing times without kafka removing the consumer from the consumer group without having to set the session timeout to the worst case processing time. Is that right?

tvoinarovskyi · 2018-05-12T16:03:36Z

@sanfilippopablo

Yes, as of 1.4.0
Yes, your assumption is correct

vkjv · 2019-02-13T14:37:39Z

May be its quite long, but I am running into the same issue with 1.4.4 client. I am using iterator interface to receive messages, which is in a separate thread. And the processing is done in another thread. All the configuration parameters set to default except for request_timeout which is set to 60seconds.
Every 5 minutes, I get the error heartbeat expired and consumer is dead.
And messages have been lost. (I am not sure about it, but it seems so)
How do I get over it?

jeffwidman · 2019-02-27T18:05:33Z

@vkjv this is a closed issue, you are better off opening a new one.

dpkp requested review from jeffwidman and tvoinarovskyi October 18, 2017 00:18

jeffwidman mentioned this pull request Oct 19, 2017

Explicitly check for None rather than falsey #1269

Merged

jeffwidman reviewed Oct 19, 2017

View reviewed changes

dpkp force-pushed the KAFKA_3888_heartbeat_thread branch from 6cdaf9f to 50640a3 Compare October 20, 2017 02:42

jeffwidman mentioned this pull request Oct 20, 2017

Bump KafkaConsumer's request_timeout to 305000ms #1002

Closed

tvoinarovskyi mentioned this pull request Oct 21, 2017

Move callback processing from BrokerConnection to KafkaClient #1258

Merged

dpkp force-pushed the KAFKA_3888_heartbeat_thread branch from 50640a3 to ea6de4c Compare October 21, 2017 16:31

tvoinarovskyi approved these changes Oct 22, 2017

View reviewed changes

This was referenced Oct 22, 2017

Auto-commit semantics #722

Closed

KafkaConsumer using group_id with manually assigned partitions can raise unexpected IllegalStateError #1112

Closed

dpkp mentioned this pull request Oct 24, 2017

KAFKA-3949: Fix race condition between group rebalance and metadata update #1241

Closed

jeffwidman mentioned this pull request Dec 8, 2017

Failures in struct packing should fail and not hang forever #1319

Open

dpkp force-pushed the KAFKA_3888_heartbeat_thread branch from 9d2087e to 276c2a2 Compare December 20, 2017 18:28

dpkp added 4 commits December 21, 2017 11:39

KAFKA-3888: use background thread for consumer heartbeats

7fa4560

Updates from review

21484ca

Whitespace fixup

7a34cce

Handle no group, no broker support, and/or manually-assigned partitio…

1247a6f

…ns in coordinator.poll()

dpkp added 9 commits December 21, 2017 11:39

Fix consumer iterator internal timeout to just check when coordinator…

4209b88

… needs next poll

coordinator.poll should not use join group with 0.8.2 brokers

3432a0b

Drop heartbeat_thread reference after close

e579490

Add simple test for consumer heartbeat thread

ccff76d

Handle session_timeout_ms / max_poll_interval_ms defaults differently…

5b2a483

… for old brokers

Fixup KafkaConfigurationError

2118555

Fixup comments

a9fdfb0

fix tests

e039108

Fix lookup future is_done check

a709cc5

dpkp force-pushed the KAFKA_3888_heartbeat_thread branch from 276c2a2 to a709cc5 Compare December 21, 2017 19:39

dpkp merged commit ad024d1 into master Dec 21, 2017

jeffwidman deleted the KAFKA_3888_heartbeat_thread branch December 21, 2017 22:58

pablasso mentioned this pull request Mar 6, 2018

KIP-62 / KAFKA-3888: Allow consumer to send heartbeats from a background thread #948

Closed

dpkp mentioned this pull request Mar 8, 2018

Client always chooses IPv6, even if connection fails #838

Closed

KAFKA-3888 Use background thread to process consumer heartbeats #1266

KAFKA-3888 Use background thread to process consumer heartbeats #1266

Conversation

dpkp commented Oct 18, 2017

jeffwidman left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dpkp Oct 21, 2017 • edited Loading

Choose a reason for hiding this comment

jeffwidman Oct 22, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dpkp Oct 21, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tvoinarovskyi commented Oct 21, 2017

tvoinarovskyi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tvoinarovskyi commented Oct 22, 2017

dpkp commented Oct 22, 2017

tvoinarovskyi commented Oct 23, 2017

tvoinarovskyi commented Oct 23, 2017

dpkp commented Oct 23, 2017

jeffwidman commented Nov 7, 2017

dpkp commented Nov 7, 2017 • edited Loading

jeffwidman commented Nov 8, 2017

tvoinarovskyi commented Dec 17, 2017

dpkp commented Dec 17, 2017 via email

dpkp commented Dec 20, 2017

jeffwidman commented Dec 21, 2017

everpcpc commented Dec 28, 2017

sanfilippopablo commented Apr 12, 2018

tvoinarovskyi commented May 12, 2018

vkjv commented Feb 13, 2019 • edited Loading

jeffwidman commented Feb 27, 2019

dpkp Oct 21, 2017 •

edited

Loading

jeffwidman Oct 22, 2017 •

edited

Loading

dpkp Oct 21, 2017 •

edited

Loading

dpkp commented Nov 7, 2017 •

edited

Loading

vkjv commented Feb 13, 2019 •

edited

Loading