[BUG]: DFP Unable to cast Python instance of type <class 'pandas.core.frame.DataFrame'> to C++ type 'std::shared_ptr<morpheus::MessageMeta>'

[dagardner-nv]

Version

25.02

Which installation method(s) does this occur on?

Docker, Source

Describe the bug.

DFP pipelines failing with a RuntimeError. The DFPSplitUsersStage should be responsible and capable of handling this conversion.

Minimum reproducible example

python dfp_duo_pipeline.py --train_users generic --start_time "2022-08-01" --input_file="../../data/dfp/duo-training-data/*.json"

Relevant log output

Click here to see error details

E20250113 15:12:34.598865 140589144065728 context.cpp:124] /linear_segment_0/dfp-file-to-df-2; rank: 0; size: 1; tid: 140589144065728: set_exception issued; issuing kill to current runnable. Exception msg: Unable to cast Python instance of type <class 'pandas.core.frame.DataFrame'> to C++ type 'std::shared_ptrmorpheus::MessageMeta'

E20250113 15:12:34.612554 140608070764352 runner.cpp:189] Runner::await_join - an exception was caught while awaiting on one or more contexts/instances - rethrowing

E20250113 15:12:34.616504 140608070764352 segment_instance.cpp:273] segment::SegmentInstance - an exception was caught while awaiting on one or more nodes - rethrowing

E20250113 15:12:34.616536 140608070764352 service.cpp:224] Service[segment::SegmentInstance]: caught exception in service_await_join: Unable to cast Python instance of type <class 'pandas.core.frame.DataFrame'> to C++ type 'std::shared_ptrmorpheus::MessageMeta'

E20250113 15:12:34.616563 140608070764352 pipeline_instance.cpp:230] pipeline::PipelineInstance - an exception was caught while awaiting on segments - rethrowing

E20250113 15:12:34.616583 140608070764352 service.cpp:224] Service[pipeline::PipelineInstance]: caught exception in service_await_join: Unable to cast Python instance of type <class 'pandas.core.frame.DataFrame'> to C++ type 'std::shared_ptrmorpheus::MessageMeta'

E20250113 15:12:34.616633 140608070764352 service.cpp:224] Service[pipeline::Manager]: caught exception in service_await_join: Unable to cast Python instance of type <class 'pandas.core.frame.DataFrame'> to C++ type 'std::shared_ptrmorpheus::MessageMeta'

E20250113 15:12:34.616650 140608070764352 service.cpp:224] Service[ExecutorDefinition]: caught exception in service_await_join: Unable to cast Python instance of type <class 'pandas.core.frame.DataFrame'> to C++ type 'std::shared_ptrmorpheus::MessageMeta'

Exception occurred in pipeline. Rethrowing

Traceback (most recent call last):

File "/home/dagardner/work/m2/python/morpheus/morpheus/pipeline/pipeline.py", line 408, in post_start

await executor.join_async()

RuntimeError: Unable to cast Python instance of type <class 'pandas.core.frame.DataFrame'> to C++ type 'std::shared_ptrmorpheus::MessageMeta'

Traceback (most recent call last):

File "/home/dagardner/work/m2/examples/digital_fingerprinting/production/dfp_duo_pipeline.py", line 376, in 

run_pipeline(obj={}, auto_envvar_prefix='DFP', show_default=True, prog_name="dfp")

File "/home/dagardner/work/conda/envs/m2/lib/python3.10/site-packages/click/core.py", line 1161, in call

return self.main(*args, **kwargs)

File "/home/dagardner/work/conda/envs/m2/lib/python3.10/site-packages/click/core.py", line 1082, in main

rv = self.invoke(ctx)

File "/home/dagardner/work/conda/envs/m2/lib/python3.10/site-packages/click/core.py", line 1443, in invoke

return ctx.invoke(self.callback, **ctx.params)

File "/home/dagardner/work/conda/envs/m2/lib/python3.10/site-packages/click/core.py", line 788, in invoke

return __callback(*args, **kwargs)

File "/home/dagardner/work/m2/examples/digital_fingerprinting/production/dfp_duo_pipeline.py", line 371, in run_pipeline

pipeline.run()

File "/home/dagardner/work/m2/python/morpheus/morpheus/pipeline/pipeline.py", line 654, in run

asyncio.run(self.run_async())

File "/home/dagardner/work/conda/envs/m2/lib/python3.10/asyncio/runners.py", line 44, in run

return loop.run_until_complete(main)

File "/home/dagardner/work/conda/envs/m2/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete

return future.result()

File "/home/dagardner/work/m2/python/morpheus/morpheus/pipeline/pipeline.py", line 635, in run_async

await self.join()

File "/home/dagardner/work/m2/python/morpheus/morpheus/pipeline/pipeline.py", line 452, in join

await self._post_start_future

File "/home/dagardner/work/m2/python/morpheus/morpheus/pipeline/pipeline.py", line 408, in post_start

await executor.join_async()

RuntimeError: Unable to cast Python instance of type <class 'pandas.core.frame.DataFrame'> to C++ type 'std::shared_ptrmorpheus::MessageMeta'

Full env printout

Click here to see environment details

[Paste the results of print_env.sh here, it will be hidden by default]

Other/Misc.

No response

Code of Conduct

• I agree to follow Morpheus' Code of Conduct
• I have searched the open bugs and have found no duplicates for this bug report

[dagardner-nv]

Appears to be caused by #1908, problem goes away if I revert 747680c

[dagardner-nv]

Looking just at the dfp_duo_pipeline pipeline, it's the first MonitorStage instance causing the trouble, commenting that out avoids the issue.