2
0
mirror of https://github.com/xcat2/confluent.git synced 2024-11-26 19:40:12 +00:00

Defer txcount increment until after potential failure

This avoids the txcount incrementing when no transaction actually
occurs.
This commit is contained in:
Jarrod Johnson 2018-07-12 08:41:07 -04:00
parent 87d00b7447
commit bd2a3f14e6

View File

@ -239,11 +239,11 @@ def exec_on_leader(function, *args):
def exec_on_followers(fnname, *args):
global _txcount
_txcount += 1
if len(cfgstreams) < (len(_cfgstore['collective']) // 2):
# the leader counts in addition to registered streams
raise exc.DegradedCollective()
pushes = eventlet.GreenPool()
_txcount += 1
payload = cPickle.dumps({'function': fnname, 'args': args,
'txcount': _txcount})
for res in pushes.starmap(