Our Actions version is v0.0.9 and Datahub version is v0.9.3

and this is the config for the pipeline that we've observed to stop a few times now

name: "dataprocessinstance_mcle_pipeline"
source:
  type: "kafka"
  config:
    connection:
      bootstrap: ${KAFKA_BOOTSTRAP_SERVER:-localhost:9092}
      consumer_config:
        security.protocol: ${KAFKA_PROPERTIES_SECURITY_PROTOCOL:-PLAINTEXT}
      schema_registry_url: ${SCHEMA_REGISTRY_URL:-<http://localhost:8081>}
    topic_routes:
      mcl: ${METADATA_CHANGE_LOG_VERSIONED_TOPIC_NAME:-MetadataChangeLog_Versioned_v1}
      pe: ${PLATFORM_EVENT_TOPIC_NAME:-PlatformEvent_v1}
filter:
  event_type: "MetadataChangeLogEvent_v1"
  event:
    entityType: "dataProcessInstance"
    changeType: "UPSERT"
    aspectName:
      - "dataProcessInstanceProperties"
action:
  type: "dataprocessinstance_mcle_action:DataProcessInstanceMetadataChangeLogEventAction"
datahub:
  server: "${DATAHUB_GMS_PROTOCOL:-http}://${DATAHUB_GMS_HOST:-localhost}:${DATAHUB_GMS_PORT:-8080}"
  extra_headers:
    Authorization: "Basic ${DATAHUB_SYSTEM_CLIENT_ID:-__datahub_system}:${DATAHUB_SYSTEM_CLIENT_SECRET:-JohnSnowKnowsNothing}"

also I'm not sure if this detail is important but after I restart the pod, the pipeline just picks up where it left off

@bulky-soccer-26729 so I have seen the issue happen again. I can't find the logs but I can confirm that right before the pipeline stopped processing, it wrote an event to the

failed_events.log

, although it is not obvious why this event failed. This is weird because while I didn't explicitly configure the

failure_mode

the documentation states that it should default to

CONTINUE

, but it seems to be acting like it was configured to

THROW

I'm going to try to explicitly configure the

failure_mode

and fix our log retention so I can debug further but if you figure anything out please let me know

hey @bulky-soccer-26729 I have some updates for this issue. I was able to isolate the issue to this error message that happens when the pipeline stops processing

Exception in thread Thread-6 (run_pipeline):
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/threading.py", line 1016, in _bootstrap_inner
    self.run()
  File "/usr/local/lib/python3.10/threading.py", line 953, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/lib/python3.10/site-packages/datahub_actions/pipeline/pipeline_manager.py", line 42, in run_pipeline
    pipeline.run()
  File "/usr/local/lib/python3.10/site-packages/datahub_actions/pipeline/pipeline.py", line 166, in run
    for enveloped_event in enveloped_events:
  File "/usr/local/lib/python3.10/site-packages/datahub_actions/plugin/source/kafka/kafka_event_source.py", line 154, in events
    msg = self.consumer.poll(timeout=2.0)
  File "/usr/local/lib/python3.10/site-packages/confluent_kafka/deserializing_consumer.py", line 131, in poll
    raise ConsumeError(msg.error(), kafka_message=msg)
KafkaError{code=_MAX_POLL_EXCEEDED,val=-147,str="Application maximum poll interval (10000ms) exceeded by 62ms"}

Does this look familiar to you?

actually I see that this was improved in a recent change so maybe all I need to do is upgrade my actions container version https://github.com/acryldata/datahub-actions/pull/74

but great call - that PR actually might be just what you need

I would suggest upgrading and see if that fixes your situation!