[Core feature] Add Kafka support for workflow notifications

[devictr]

Motivation: Why do you think this is important?

Notifications only support AWS and GCP for the moment. My company is on Azure and would benefit from having notifications as well.
It seems like Flyte uses gizmo for this, which only supports AWS, GCP, Kafka and HTTP.

Adding Kafka support would benefit Azure deployments, but also on-premise or other deployments that already have Kafka setups internally.

Goal: What should the final outcome look like, ideally?

I want to be able to configure Kafka for workflow notifications.

Describe alternatives you've considered

Alternatives: HTTP support from gizmo. This feature doesn't exclude HTTP support from being added in the future.

Propose: Link/Inline OR Additional context

Code:

flyte/flyteadmin/pkg/async/notifications/factory.go

Lines 95 to 147 in b30cb72

	switch config.Type {
	case common.AWS:
	sqsConfig := gizmoAWS.SQSConfig{
	QueueName: config.NotificationsProcessorConfig.QueueName,
	QueueOwnerAccountID: config.NotificationsProcessorConfig.AccountID,
	// The AWS configuration type uses SNS to SQS for notifications.
	// Gizmo by default will decode the SQS message using Base64 decoding.
	// However, the message body of SQS is the SNS message format which isn't Base64 encoded.
	ConsumeBase64: &enable64decoding,
	}
	if config.AWSConfig.Region != "" {
	sqsConfig.Region = config.AWSConfig.Region
	} else {
	sqsConfig.Region = config.Region
	}
	var err error
	err = async.Retry(reconnectAttempts, reconnectDelay, func() error {
	sub, err = gizmoAWS.NewSubscriber(sqsConfig)
	if err != nil {
	logger.Warnf(context.TODO(), "Failed to initialize new gizmo aws subscriber with config [%+v] and err: %v", sqsConfig, err)
	}
	return err
	})
	if err != nil {
	panic(err)
	}
	emailer = GetEmailer(config, scope)
	return implementations.NewProcessor(sub, emailer, scope)
	case common.GCP:
	projectID := config.GCPConfig.ProjectID
	subscription := config.NotificationsProcessorConfig.QueueName
	var err error
	err = async.Retry(reconnectAttempts, reconnectDelay, func() error {
	sub, err = gizmoGCP.NewSubscriber(context.TODO(), projectID, subscription)
	if err != nil {
	logger.Warnf(context.TODO(), "Failed to initialize new gizmo gcp subscriber with config [ProjectID: %s, Subscription: %s] and err: %v", projectID, subscription, err)
	}
	return err
	})
	if err != nil {
	panic(err)
	}
	emailer = GetEmailer(config, scope)
	return implementations.NewGcpProcessor(sub, emailer, scope)
	case common.Sandbox:
	emailer = GetEmailer(config, scope)
	return implementations.NewSandboxProcessor(msgChan, emailer)
	case common.Local:
	fallthrough
	default:
	logger.Infof(context.Background(),
	"Using default noop notifications processor implementation for config type [%s]", config.Type)

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

• Yes

Have you read the Code of Conduct?

• Yes

[Future-Outlier]

I added the sandbox email publisher 3 months ago.
If you need help, feel free to collaborate with me.
PR Link: flyteorg/flyteadmin#595

[devictr]

Thanks @Future-Outlier!

[devictr]

@pingsutw, any preference as to where to store Kafka offsets? Since this is flyteadmin, maybe I can create a table in postgres for that?

[devictr]

Actually, seems like we use Sarama behind the scenes, which supports auto commit and is enabled by default: https://github.com/IBM/sarama/blob/main/config.go#L427

[davidmirror-ops]

with gizmo on maintenance mode do we still consider following their http route?

[eapolinario]

We talked offline about this and the conclusion is that we're not married to gizmo. We'd welcome a PR that brings in another dependency to implement Azure workflow notifications.

[Sovietaced]

We have an implementation of this internally that we could potentially upstream.