jan/openreplay
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(connector): Fixed small issues and added checkpoint method (#1251)

Browse source 
* fix(connector): fixed bug of cache dict size error

* fix(connector): Added method to save state in s3 for redshift if sigterm arise

* fix(connector): Added exit signal handler and checkpoint method

* Added sslmode selection for connection to database, added use_ssl parameter for S3 connection

* fix(connector): Handle error when broken session_id
This commit is contained in:
MauricioGarciaS 2023-05-31 11:42:49 +02:00 • committed by GitHub
parent d958daa61e
commit 164232377d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 8 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
ee/connectors/consumer_async.py
View file

@ -22,7 +22,12 @@ def process_message(msg, codec, sessions, batch, sessions_batch, interesting_ses
if msg is None:
return
messages = codec.decode_detailed(msg.value())
session_id = codec.decode_key(msg.key())
try:
session_id = codec.decode_key(msg.key())
except Exception as e:
print('[WARN] Broken sessionid')
print(e)
return
if messages is None:
print('-')
return

3
ee/connectors/msgcodec/msgcodec.py
View file

@ -43,7 +43,8 @@ class MessageCodec(Codec):
try:
decoded = int.from_bytes(b, "little", signed=False)
except Exception as e:
raise UnicodeDecodeError(f"Error while decoding message key (SessionID) from {b}\n{e}")
print(f"Error while decoding message key (SessionID) from {b}\n{e}")
raise e
return decoded
def decode_detailed(self, b: bytes) -> List[Message]: