rh-marketingops/fluvii

A kafka streams client library built on confluent-kafka-python

fluvii's Issues

Update Setuptools name parameter for fluvii

suggestion instead of setuptools.setup(name="gtfo",) which setuptools.setup(name="fluvii",)

Update document with FluviiAppConfig examples

Update readme document with more detailed examples of configuration FluviiAppConfig

v0.2.10 - Gracefully handle "Erroneous state" Kafka error

App was running fluvii==0.2.10

"2022-11-14T14:10:13.930345965+00:00 stderr F ",,
"2022-11-14T14:10:13.930332630+00:00 stderr P : KafkaError{code=_STATE,val=-172,str=""Unable to produce message: Local: Erroneous state""}",,
"2022-11-14T14:10:13.930324320+00:00 stderr P cimpl.KafkaException",,
"2022-11-14T14:10:13.930311558+00:00 stderr F     super(SerializingProducer, self).produce(topic, value, key,",,
"2022-11-14T14:10:13.929698863+00:00 stderr F   File ""/opt/app-root/lib64/python3.8/site-packages/confluent_kafka/serializing_producer.py"", line 176, in produce",,
"2022-11-14T14:10:13.929698863+00:00 stderr F     self._producer.produce(**produce_dict)",,
"2022-11-14T14:10:13.929640738+00:00 stderr F   File ""/opt/app-root/lib64/python3.8/site-packages/fluvii/producer/producer.py"", line 137, in produce",,
"2022-11-14T14:10:13.929640738+00:00 stderr F     super().produce(value, key=key, topic=topic, headers=headers, partition=partition, message_passthrough=message_passthrough)",,
"2022-11-14T14:10:13.929566016+00:00 stderr F   File ""/opt/app-root/lib64/python3.8/site-packages/fluvii/producer/producer.py"", line 186, in produce",,
"2022-11-14T14:10:13.929566016+00:00 stderr F     self.producer.produce(value, message_passthrough=self.message, **producer_kwargs)",,
"2022-11-14T14:10:13.929508168+00:00 stderr F   File ""/opt/app-root/lib64/python3.8/site-packages/fluvii/transaction/transaction.py"", line 99, in produce",,
"2022-11-14T14:10:13.929508168+00:00 stderr F     transaction.produce(dict(",,
"2022-11-14T14:10:13.929459559+00:00 stderr F   File ""/opt/app-root/src/dept_role_persona_deriver/helper_funcs/helpers.py"", line 35, in produce_message",,
"2022-11-14T14:10:13.929459559+00:00 stderr F     produce_message(transaction, msg)",,
"2022-11-14T14:10:13.929415759+00:00 stderr F   File ""/opt/app-root/src/dept_role_persona_deriver/deriver_app.py"", line 15, in dept_role_persona_deriver_loop",,
"2022-11-14T14:10:13.929415759+00:00 stderr F     self._app_function(self.transaction, *self._app_function_arglist)",,
"2022-11-14T14:10:13.929362491+00:00 stderr F   File ""/opt/app-root/lib64/python3.8/site-packages/fluvii/fluvii_app/fluvii_app.py"", line 104, in _handle_message",,
"2022-11-14T14:10:13.929362491+00:00 stderr F     self._handle_message(**kwargs)",,
"2022-11-14T14:10:13.929308054+00:00 stderr F   File ""/opt/app-root/lib64/python3.8/site-packages/fluvii/fluvii_app/fluvii_app.py"", line 120, in _app_batch_run_loop",,
"2022-11-14T14:10:13.929308054+00:00 stderr F     self._app_batch_run_loop(**kwargs)",,
"2022-11-14T14:10:13.929250609+00:00 stderr F   File ""/opt/app-root/lib64/python3.8/site-packages/fluvii/fluvii_app/fluvii_app.py"", line 130, in _app_batch_run_loop",,
"2022-11-14T14:10:13.929250609+00:00 stderr F     self._app_batch_run_loop(**kwargs)",,
"2022-11-14T14:10:13.929176394+00:00 stderr F   File ""/opt/app-root/lib64/python3.8/site-packages/fluvii/fluvii_app/fluvii_app.py"", line 152, in _run",,
"2022-11-14T14:10:13.929176394+00:00 stderr F     log_and_raise_error(self.metrics_manager, e)",,
"2022-11-14T14:10:13.929103071+00:00 stderr F   File ""/opt/app-root/lib64/python3.8/site-packages/fluvii/fluvii_app/fluvii_app.py"", line 158, in _run",,
"2022-11-14T14:10:13.929081098+00:00 stderr F     self._run(**kwargs)",,
"2022-11-14T14:10:13.929007522+00:00 stderr F   File ""/opt/app-root/lib64/python3.8/site-packages/fluvii/fluvii_app/fluvii_app.py"", line 181, in run",,
"2022-11-14T14:10:13.929007522+00:00 stderr F     super().run()",,
"2022-11-14T14:10:13.928953723+00:00 stderr F   File ""/opt/app-root/lib64/python3.8/site-packages/nubium_utils/confluent_utils/gtfo/fluvii_extensions/nubium/nubium_app.py"", line 88, in run",,
"2022-11-14T14:10:13.928945141+00:00 stderr F     self._app.run()",,
"2022-11-14T14:10:13.928864971+00:00 stderr F   File ""/opt/app-root/lib64/python3.8/site-packages/nubium_utils/confluent_utils/gtfo/gtfo_app.py"", line 61, in run",,
"2022-11-14T14:10:13.928854467+00:00 stderr F     dept_role_persona_deriver().run()",,
"2022-11-14T14:10:13.928503124+00:00 stderr F   File ""app.py"", line 3, in <module>",,
"2022-11-14T14:10:13.928503124+00:00 stderr F Traceback (most recent call last):",,
"2022-11-14T14:10:13.924633542+00:00 stdout F PID 1 - 2022-11-14 14:10:13,924 - fluvii.fluvii_app.fluvii_app - INFO: Performing graceful teardown of producer and/or consumer...",,
"2022-11-14T14:10:13.924176777+00:00 stdout F PID 1 - 2022-11-14 14:10:13,924 - fluvii.producer.producer - INFO: Adding serializer for producer topic PeopleStream_DataWashingMachine_AddressMsa",,
"2022-11-14T14:10:13.915735619+00:00 stdout F PID 1 - 2022-11-14 14:10:13,915 - fluvii.producer.producer - INFO: Adding schema for topic PeopleStream_DataWashingMachine_AddressMsa",,
"2022-11-14T14:10:13.411524089+00:00 stderr F %4|1668435013.411|GETPID|rdkafka#producer-3| [thrd:main]: Failed to acquire transactional PID from broker TxnCoordinator/0: Broker: Producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing: retrying",,
"2022-11-14T14:10:13.260472107+00:00 stdout F PID 1 - 2022-11-14 14:10:13,260 - fluvii.producer.producer - INFO: Producer initialized successfully!",,
"2022-11-14T14:10:13.100155176+00:00 stdout F PID 1 - 2022-11-14 14:10:13,100 - fluvii.producer.producer - INFO: Initializing producer...",,
"2022-11-14T14:10:13.100068760+00:00 stdout F PID 1 - 2022-11-14 14:10:13,100 - fluvii.transaction.transaction - INFO: Failed to abort transaction: KafkaError{code=_STATE,val=-172,str=""Operation not valid in state BeginCommit""}",,
"2022-11-14T14:10:13.099982720+00:00 stdout F PID 1 - 2022-11-14 14:10:13,099 - fluvii.transaction.transaction - INFO: Aborting transaction.",,
"2022-11-14T14:10:13.099909260+00:00 stdout F PID 1 - 2022-11-14 14:10:13,099 - fluvii.consumer.consumer - INFO: Reversing topic PeopleStream_DataWashingMachine_DeptJobrolePersona partition 8 back to offset 2429011",,
"2022-11-14T14:10:13.099829074+00:00 stdout F PID 1 - 2022-11-14 14:10:13,099 - fluvii.consumer.consumer - INFO: Reversing topic PeopleStream_DataWashingMachine_DeptJobrolePersona partition 10 back to offset 2432274",,
"2022-11-14T14:10:13.099711589+00:00 stdout F PID 1 - 2022-11-14 14:10:13,099 - fluvii.consumer.consumer - INFO: Rolling back consumer state to earliest non-committed offset(s)...",,
"2022-11-14T14:10:13.099667565+00:00 stdout F PID 1 - 2022-11-14 14:10:13,099 - fluvii.fluvii_app.fluvii_app - INFO: App is shutting down...",,
"2022-11-14T14:10:08.094401919+00:00 stdout F PID 1 - 2022-11-14 14:10:08,094 - fluvii.fluvii_app.fluvii_app - ERROR: <class 'cimpl.KafkaException'>, KafkaError{code=_STATE,val=-172,str=""Unable to produce message: Local: Erroneous state""}",,
"2022-11-14T14:10:08.091600889+00:00 stdout F PID 1 - 2022-11-14 14:10:08,091 - fluvii.fluvii_app.fluvii_app - INFO: Graceful transaction failure; retrying commit...",,
"2022-11-14T14:10:08.091549473+00:00 stdout F PID 1 - 2022-11-14 14:10:08,091 - fluvii.transaction.transaction - INFO: KafkaException: is retriable? - True, should abort? - False",,
"2022-11-14T14:10:08.091498570+00:00 stdout F PID 1 - 2022-11-14 14:10:08,091 - fluvii.transaction.transaction - ERROR: <built-in method code of cimpl.KafkaError object at 0x7f1117374430>",,
"2022-11-14T14:10:08.091451467+00:00 stdout F PID 1 - 2022-11-14 14:10:08,091 - fluvii.transaction.transaction - ERROR: KafkaError{code=_TIMED_OUT,val=-185,str=""Failed to flush all outstanding messages within the transaction timeout: 4 message(s) remaining""}",,

Delay each consumption process

Hey! first, thanks for sharing this project!!

Well, I need to consume an event and call an external API which has a limit of 1 request per second. It's possible to consume only one message per time and add a delay of 1 second to each consumption process?

v0.2.11 - Non-graceful recovery after a transaction timeout

In Fluvii version 0.2.11, it appears that the producer needs an "epoch refresh" after a timeout exception.

PID 1 - 2022-11-19 17:17:43,092 - eloqua_bulk_batcher_controller.helper_funcs.status_methods - INFO: PeopleStream_PeopleStreamFromEloqua_ContactRetriever__2022-11-19T17:13:39Z__2022-11-19T17:17:14Z - Now waiting for sync creation, population, and record count...
PID 1 - 2022-11-19 17:17:43,092 - fluvii.producer.producer - INFO: Producing message with guid 13639a82-682e-11ed-90a6-0a58ac147930
PID 1 - 2022-11-19 17:17:43,092 - fluvii.consumer.consumer - INFO: Consumption attempts for this batch are finished.
%6|1668878263.093|FAIL|rdkafka#producer-1| [thrd:TxnCoordinator]: TxnCoordinator: broker-1-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443: Disconnected (after 559819ms in state UP, 1 identical error(s) suppressed)
%6|1668878263.161|FAIL|rdkafka#producer-1| [thrd:sasl_ssl://broker-4-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcl]: sasl_ssl://broker-4-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/4: Disconnected (after 559889ms in state UP, 1 identical error(s) suppressed)
%6|1668878263.594|FAIL|rdkafka#producer-1| [thrd:sasl_ssl://mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:4]: sasl_ssl://mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/bootstrap: Disconnected (after 560321ms in state UP, 1 identical error(s) suppressed)
%6|1668878264.095|FAIL|rdkafka#producer-1| [thrd:sasl_ssl://broker-2-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcl]: sasl_ssl://broker-2-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/2: Disconnected (after 560820ms in state UP, 1 identical error(s) suppressed)
%6|1668878292.077|FAIL|rdkafka#consumer-2| [thrd:sasl_ssl://broker-3-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcl]: sasl_ssl://broker-3-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/3: Disconnected (after 214671ms in state UP)
%6|1668878292.081|FAIL|rdkafka#consumer-2| [thrd:sasl_ssl://broker-4-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcl]: sasl_ssl://broker-4-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/4: Disconnected (after 244698ms in state UP, 1 identical error(s) suppressed)
%6|1668878292.084|FAIL|rdkafka#consumer-2| [thrd:sasl_ssl://broker-5-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcl]: sasl_ssl://broker-5-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/5: Disconnected (after 244710ms in state UP, 1 identical error(s) suppressed)
%6|1668878292.087|FAIL|rdkafka#consumer-2| [thrd:sasl_ssl://broker-2-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcl]: sasl_ssl://broker-2-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/2: Disconnected (after 244701ms in state UP, 1 identical error(s) suppressed)
%6|1668878292.089|FAIL|rdkafka#consumer-2| [thrd:sasl_ssl://broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcl]: sasl_ssl://broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/0: Disconnected (after 244710ms in state UP, 1 identical error(s) suppressed)
%6|1668878292.092|FAIL|rdkafka#consumer-2| [thrd:sasl_ssl://broker-1-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcl]: sasl_ssl://broker-1-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/1: Disconnected (after 244704ms in state UP, 1 identical error(s) suppressed)
%6|1668878292.094|FAIL|rdkafka#consumer-2| [thrd:GroupCoordinator]: GroupCoordinator: broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443: Disconnected (after 194267ms in state UP)
%4|1668878292.095|FAIL|rdkafka#consumer-2| [thrd:sasl_ssl://mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:4]: sasl_ssl://mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/bootstrap: Disconnected (after 1142751ms in state UP, 1 identical error(s) suppressed)
PID 1 - 2022-11-19 17:18:43,092 - fluvii.transaction.transaction - ERROR: KafkaError{code=_TIMED_OUT,val=-185,str="Transactional operation timed out"}
PID 1 - 2022-11-19 17:18:43,093 - fluvii.transaction.transaction - ERROR: <built-in method code of cimpl.KafkaError object at 0x7f88af1d0c10>
PID 1 - 2022-11-19 17:18:43,093 - fluvii.transaction.transaction - INFO: KafkaException: is retriable? - True, should abort? - False
PID 1 - 2022-11-19 17:18:43,093 - fluvii.fluvii_app.fluvii_app - INFO: Graceful transaction failure; retrying commit...
PID 1 - 2022-11-19 17:18:43,093 - fluvii.fluvii_app.fluvii_app - INFO: Consuming 1 messages over 10 seconds...
%5|1668878323.176|REQTMOUT|rdkafka#producer-1| [thrd:sasl_ssl://broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcl]: sasl_ssl://broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/0: Timed out MetadataRequest in flight (after 60083ms, timeout #0)
%5|1668878323.176|REQTMOUT|rdkafka#producer-1| [thrd:sasl_ssl://broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcl]: sasl_ssl://broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/0: Timed out ProduceRequest in flight (after 60016ms, timeout #1)
%5|1668878323.176|REQTMOUT|rdkafka#producer-1| [thrd:sasl_ssl://broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcl]: sasl_ssl://broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/0: Timed out MetadataRequest in flight (after 60015ms, timeout #2)
%5|1668878323.176|REQTMOUT|rdkafka#producer-1| [thrd:sasl_ssl://broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcl]: sasl_ssl://broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/0: Timed out FindCoordinatorRequest in flight (after 59979ms, timeout #3)
%5|1668878323.176|REQTMOUT|rdkafka#producer-1| [thrd:sasl_ssl://broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcl]: sasl_ssl://broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/0: Timed out TxnOffsetCommitRequest in flight (after 59946ms, timeout #4)
%4|1668878323.177|REQTMOUT|rdkafka#producer-1| [thrd:sasl_ssl://broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcl]: sasl_ssl://broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/0: Timed out 5 in-flight, 0 retry-queued, 0 out-queue, 0 partially-sent requests
%3|1668878323.177|FAIL|rdkafka#producer-1| [thrd:sasl_ssl://broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcl]: sasl_ssl://broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/0: 5 request(s) timed out: disconnect (after 319861ms in state UP)
%3|1668878323.180|TXNERR|rdkafka#producer-1| [thrd:sasl_ssl://broker-0-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcl]: Current transaction failed in state InTransaction: 1 message(s) timed out on NubiumIntegrations_Eloqua_EbbWorkerTasks [10] (_TIMED_OUT, requires epoch bump)
%6|1668878323.184|FAIL|rdkafka#producer-1| [thrd:sasl_ssl://broker-5-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcl]: sasl_ssl://broker-5-mode-ps-cbi-g--tbjmfoubd-hvg.bf2.kafka.rhcloud.com:443/5: Disconnected (after 319824ms in state UP, 1 identical error(s) suppressed)
%4|1668878323.295|GETPID|rdkafka#producer-1| [thrd:main]: Failed to acquire transactional PID from broker TxnCoordinator/1: Broker: Producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing: retrying
%1|1668878323.798|TXNERR|rdkafka#producer-1| [thrd:main]: Fatal transaction error: Producer fenced by newer instance: Failed to acquire transactional PID from broker TxnCoordinator/1: Broker: There is a newer producer with the same transactionalId which fences the current one (_FENCED)

Schema registry not found - invalid url?

I am using v0.2.14 and get the following error:

No connection adapters were found for 'httphttp://kafka-schema-registry:8081/schemas/ids/10'

Environment:

  TZ: Europe/Zurich
  FLUVII_KAFKA_BOOTSTRAP_SERVERS: "kafka1:19092"
  FLUVII_SCHEMA_REGISTRY_URL: "http://kafka-schema-registry:8081"

Traceback:

Traceback (most recent call last):
File "/usr/local/lib/python3.10/site-packages/confluent_kafka/deserializing_consumer.py", line 137, in poll
value = self._value_deserializer(value, ctx)
File "/usr/local/lib/python3.10/site-packages/confluent_kafka/schema_registry/avro.py", line 357, in call
schema = self._registry.get_schema(schema_id)
File "/usr/local/lib/python3.10/site-packages/confluent_kafka/schema_registry/schema_registry_client.py", line 368, in get_schema
response = self._rest_client.get('schemas/ids/{}'.format(schema_id))
File "/usr/local/lib/python3.10/site-packages/confluent_kafka/schema_registry/schema_registry_client.py", line 124, in get
return self.send_request(url, method='GET', query=query)
File "/usr/local/lib/python3.10/site-packages/confluent_kafka/schema_registry/schema_registry_client.py", line 167, in send_request
response = self.session.request(
File "/usr/local/lib/python3.10/site-packages/requests/sessions.py", line 587, in request
resp = self.send(prep, **send_kwargs)
File "/usr/local/lib/python3.10/site-packages/requests/sessions.py", line 695, in send
adapter = self.get_adapter(url=request.url)
File "/usr/local/lib/python3.10/site-packages/requests/sessions.py", line 792, in get_adapter
raise InvalidSchema(f"No connection adapters were found for {url!r}")
requests.exceptions.InvalidSchema: No connection adapters were found for 'httphttp://kafka-schema-registry:8081/schemas/ids/10'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "//./main.py", line 20, in
fluvii_table_app().run()
File "/usr/local/lib/python3.10/site-packages/fluvii/fluvii_app/fluvii_app.py", line 183, in run
self._run(**kwargs)
File "/usr/local/lib/python3.10/site-packages/fluvii/fluvii_app/fluvii_app.py", line 160, in _run
log_and_raise_error(self.metrics_manager, e)
File "/usr/local/lib/python3.10/site-packages/fluvii/fluvii_app/fluvii_app.py", line 154, in _run
self._app_batch_run_loop(**kwargs)
File "/usr/local/lib/python3.10/site-packages/fluvii/fluvii_app/fluvii_table_app.py", line 38, in _app_batch_run_loop
super()._app_batch_run_loop(**kwargs)
File "/usr/local/lib/python3.10/site-packages/fluvii/fluvii_app/fluvii_app.py", line 122, in _app_batch_run_loop
self._handle_message(**kwargs)
File "/usr/local/lib/python3.10/site-packages/fluvii/fluvii_app/fluvii_app.py", line 105, in _handle_message
self.consume(**kwargs)
File "/usr/local/lib/python3.10/site-packages/fluvii/fluvii_app/fluvii_app.py", line 174, in consume
self.transaction.consume(**kwargs)
File "/usr/local/lib/python3.10/site-packages/fluvii/transaction/transaction.py", line 59, in consume
self.message = self.consumer.consume(**kwargs)
File "/usr/local/lib/python3.10/site-packages/fluvii/consumer/consumer.py", line 280, in consume
return super().consume(timeout=timeout)
File "/usr/local/lib/python3.10/site-packages/fluvii/consumer/consumer.py", line 120, in consume
self.message = self._poll_for_message(timeout)
File "/usr/local/lib/python3.10/site-packages/fluvii/consumer/consumer.py", line 81, in _poll_for_message
message = self._consumer.poll(timeout)
File "/usr/local/lib/python3.10/site-packages/confluent_kafka/deserializing_consumer.py", line 139, in poll
raise ValueDeserializationError(exception=se, kafka_message=msg)
confluent_kafka.error.ValueDeserializationError: KafkaError{code=_VALUE_DESERIALIZATION,val=-159,str="No connection adapters were found for 'httphttp://kafka-schema-registry:8081/schemas/ids/10'"}