[BUG]: vdb_upload pipeline failing on incompatible message type

[dagardner-nv]

Version

24.03

Which installation method(s) does this occur on?

Source

Describe the bug.

RuntimeError: The inference stage cannot handle input of <class 'morpheus._lib.messages.ControlMessage'>. Accepted input types: (<class 'morpheus.messages.multi_inference_message.MultiInferenceMessage'>,)

Minimum reproducible example

python examples/llm/main.py vdb_upload pipeline --stop_after=1024

Relevant log output

Click here to see error details

====Pipeline Pre-build====
====Pre-Building Segment: main====
Error occurred during Pipeline.build(). Exiting.
Traceback (most recent call last):
  File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/pipeline.py", line 458, in build_and_start
    self.build()
  File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/pipeline.py", line 294, in build
    self._pre_build()
  File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/pipeline.py", line 257, in _pre_build
    stage._pre_build()
  File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/source_stage.py", line 73, in _pre_build
    return super()._pre_build(do_propagate=do_propagate)
  File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/stage_base.py", line 372, in _pre_build
    dep._pre_build(do_propagate=do_propagate)
  File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/stage_base.py", line 372, in _pre_build
    dep._pre_build(do_propagate=do_propagate)
  File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/stage_base.py", line 372, in _pre_build
    dep._pre_build(do_propagate=do_propagate)
  File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/stage_base.py", line 349, in _pre_build
    self._pre_compute_schema(schema)
  File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/single_port_stage.py", line 65, in _pre_compute_schema
    raise RuntimeError((f"The {self.name} stage cannot handle input of {input_type}. "
RuntimeError: The inference stage cannot handle input of . Accepted input types: (,)
====Pipeline Complete====
Traceback (most recent call last):
  File "/home/dagardner/work/morpheus-24.03/examples/llm/main.py", line 33, in 
    run_cli()
  File "/home/dagardner/work/morpheus-24.03/examples/llm/main.py", line 29, in run_cli
    cli(obj={}, auto_envvar_prefix='MORPHEUS_LLM', show_default=True, prog_name="morpheus_llm")
  File "/home/dagardner/work/conda/envs/morpheus-24.03/lib/python3.10/site-packages/click/core.py", line 1157, in __call__
    return self.main(*args, **kwargs)
  File "/home/dagardner/work/conda/envs/morpheus-24.03/lib/python3.10/site-packages/click/core.py", line 1078, in main
    rv = self.invoke(ctx)
  File "/home/dagardner/work/conda/envs/morpheus-24.03/lib/python3.10/site-packages/click/core.py", line 1688, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/home/dagardner/work/conda/envs/morpheus-24.03/lib/python3.10/site-packages/click/core.py", line 1688, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/home/dagardner/work/conda/envs/morpheus-24.03/lib/python3.10/site-packages/click/core.py", line 1434, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/home/dagardner/work/conda/envs/morpheus-24.03/lib/python3.10/site-packages/click/core.py", line 783, in invoke
    return __callback(*args, **kwargs)
  File "/home/dagardner/work/morpheus-24.03/examples/llm/vdb_upload/run.py", line 175, in pipeline
    return _pipeline(**final_config)
  File "/home/dagardner/work/morpheus-24.03/examples/llm/vdb_upload/pipeline.py", line 104, in pipeline
    pipe.run()
  File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/pipeline.py", line 651, in run
    asyncio.run(self.run_async())
  File "/home/dagardner/work/conda/envs/morpheus-24.03/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/home/dagardner/work/conda/envs/morpheus-24.03/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete
    return future.result()
  File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/pipeline.py", line 632, in run_async
    await self.join()
  File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/pipeline.py", line 447, in join
    assert self._post_start_future is not None, "Pipeline must be started before joining"
AssertionError: Pipeline must be started before joining

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

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