GCP-based flyte notifications - message published but missing actual message

[dyu-bot]

Describe the bug

I set up GCP-based flyte notification as such:

  workflow_notifications:
    enabled: true
    config:
      notifications:
        type: gcp
        gcp:
          projectId: xxx
        publisher:
          topicName: "xxx"
        processor:
          queueName: "xxx"
        emailer:
          emailServerConfig:
            serviceName: sendgrid
            apiKeyEnvVar: xxx
          subject: "hi"
          sender:  "[email protected]"
          body: "test"

Running the sample workflow in the docs to notify upon success:

@task
def double_int_and_print(a: int) -> str:
    return str(a * 2)


@workflow
def int_doubler_wf(a: int) -> str:
    doubled = double_int_and_print(a=a)
    return doubled

wacky_int_doubler_lp = LaunchPlan.get_or_create(
    name="wacky_int_doubler",
    workflow=int_doubler_wf,
    default_inputs={"a": 4},
    notifications=[
        Slack(
            phases=[
                WorkflowExecutionPhase.SUCCEEDED,
                WorkflowExecutionPhase.ABORTED,
                WorkflowExecutionPhase.TIMED_OUT,
                WorkflowExecutionPhase.FAILED,
            ],
            recipients_email=["xxx.slack.com"],
        ),
    ],
)

The logs show that the message was published successfully:

{"json":{"exec_id":"xxx","src":"execution_manager.go:1505"},"level":"debug","msg":"publishing notifications for execution [project:\"x\" x:\"x\" name:\"x\" ] in state [SUCCEEDED] for notifications [[phases:SUCCEEDED phases:ABORTED phases:TIMED_OUT phases:FAILED slack:\u003crecipients_email:\"xxx\" \u003e ]]","ts":"x"}

and

{"json":{"exec_id":"x","src":"publisher.go:30"},"level":"debug","msg":"Publishing the following message [recipients_email:\"xxx\" sender_email:\"x\" subject_line:\"hi\" body:\"test\" ]","ts":"x"}

However, the processor is unable to pull the message.
When I send test messages to the GCP topic manually on the GCP console, the processor is able to pull properly (see below). It's only the message the Publisher sent that's not going through.

{"json":{"src":"base.go:106"},"level":"info","msg":"Started processing notifications.","ts":"2023-09-20T11:33:57Z"}
{"json":{"src":"gcp_processor.go:33"},"level":"warning","msg":"Starting GCP notifications processor","ts":"2023-09-20T11:33:57Z"}
{"json":{"src":"gcp_processor.go:47"},"level":"debug","msg":"failed to unmarshal to notification object message [asdf] with err: proto: cannot parse invalid wire-format data","ts":"2023-09-20T11:38:09Z"}
{"json":{"src":"gcp_processor.go:47"},"level":"debug","msg":"failed to unmarshal to notification object message [foo] with err: proto: cannot parse invalid wire-format data","ts":"2023-09-20T11:38:22Z"}
{"json":{"src":"gcp_processor.go:47"},"level":"debug","msg":"failed to unmarshal to notification object message [why dont you get received] with err: proto: cannot parse invalid wire-format data","ts":"2023-09-20T11:38:38Z"}
{"json":{"src":"gcp_processor.go:47"},"level":"debug","msg":"failed to unmarshal to notification object message [WHY WHY WHY] with err: proto: cannot parse invalid wire-format data","ts":"2023-09-20T11:39:52Z"}
{"json":{"src":"gcp_processor.go:47"},"level":"debug","msg":"failed to unmarshal to notification object message [PLEASE PICKUP] with err: proto: cannot parse invalid wire-format data","ts":"2023-09-20T11:40:46Z"}
{"json":{"src":"gcp_processor.go:47"},"level":"debug","msg":"failed to unmarshal to notification object message [WHY] with err: proto: cannot parse invalid wire-format data","ts":"2023-09-20T11:44:24Z"}

Interestingly enough, when I create a 2nd subscriber, i am able to pull the message that was Published.

However, the actual message is missing! Only the attribute key: flyteidl.admin.EmailNotification is received, no message body.

Expected behavior

Expected the Processor to be able to pull the message the Publisher sent.

Additional context to reproduce

No response

Screenshots

No response

Are you sure this issue hasn't been raised already?

• Yes

Have you read the Code of Conduct?

• Yes

[github-actions]

Hello 👋, this issue has been inactive for over 9 months. To help maintain a clean and focused backlog, we'll be marking this issue as stale and will engage on it to decide if it is still applicable.
Thank you for your contribution and understanding! 🙏

[davidmirror-ops]

It needs more investigation and, hopefully, an update to the docs