[CLN] Remove pulsar as a dependency to sysdb (chroma-core#1982)

## Description of changes

*Summarize the changes made by this PR.*
 - Improvements & Bug fixes
	 - This PR removes the pulsar related configuration for sysdb. 
	 - Also, removes dependency of pulsar for sysdb in Tiltfile.
 - New functionality
	 - ...

## Test plan
*How are these changes tested?*

- [ ] Tests pass locally with `pytest` for python, `yarn test` for js,
`cargo test` for rust

## Documentation Changes
*Are all docstrings for user-facing APIs updated if required? Do we need
to make documentation changes in the [docs
repository](https://github.com/chroma-core/docs)?*

Tiltfile

@@ -134,7 +134,7 @@ k8s_resource('pulsar', resource_deps=['k8s_setup', 'namespace'], labels=["infras
 k8s_resource('sysdb-migration', resource_deps=['postgres', 'namespace'], labels=["infrastructure"])
 k8s_resource('logservice-migration', resource_deps=['postgres', 'namespace'], labels=["infrastructure"])
 k8s_resource('logservice', resource_deps=['sysdb-migration'], labels=["chroma"], port_forwards='50052:50051')
-k8s_resource('sysdb', resource_deps=['pulsar', 'sysdb-migration'], labels=["chroma"], port_forwards='50051:50051')
+k8s_resource('sysdb', resource_deps=['sysdb-migration'], labels=["chroma"], port_forwards='50051:50051')
 k8s_resource('frontend-service', resource_deps=['pulsar', 'sysdb', 'logservice'],labels=["chroma"], port_forwards='8000:8000')
 k8s_resource('query-service', resource_deps=['sysdb'], labels=["chroma"])
 k8s_resource('compaction-service', resource_deps=['sysdb'], labels=["chroma"])

go/cmd/coordinator/cmd.go

@@ -41,12 +41,6 @@ func init() {
 	Cmd.Flags().IntVar(&conf.DBConfig.MaxOpenConns, "max-open-conns", 10, "MetaTable max open connections")
 	Cmd.Flags().StringVar(&conf.DBConfig.SslMode, "ssl-mode", "disable", "SSL mode for database connection")
 
-	// Pulsar
-	Cmd.Flags().StringVar(&conf.PulsarAdminURL, "pulsar-admin-url", "http://localhost:8080", "Pulsar admin url")
-	Cmd.Flags().StringVar(&conf.PulsarURL, "pulsar-url", "pulsar://localhost:6650", "Pulsar url")
-	Cmd.Flags().StringVar(&conf.PulsarTenant, "pulsar-tenant", "default", "Pulsar tenant")
-	Cmd.Flags().StringVar(&conf.PulsarNamespace, "pulsar-namespace", "default", "Pulsar namespace")
-
 	// Notification
 	Cmd.Flags().StringVar(&conf.NotificationStoreProvider, "notification-store-provider", "memory", "Notification store provider")
 	Cmd.Flags().StringVar(&conf.NotifierProvider, "notifier-provider", "memory", "Notifier provider")

go/pkg/coordinator/grpc/server.go

@@ -7,7 +7,6 @@ import (
 
 	"github.com/chroma-core/chroma/go/pkg/grpcutils"
 
-	"github.com/apache/pulsar-client-go/pulsar"
 	"github.com/chroma-core/chroma/go/pkg/coordinator"
 	"github.com/chroma-core/chroma/go/pkg/memberlist_manager"
 	"github.com/chroma-core/chroma/go/pkg/metastore/db/dao"
@@ -37,12 +36,6 @@ type Config struct {
 	NotifierProvider          string
 	NotificationTopic         string
 
-	// Pulsar config
-	PulsarAdminURL  string
-	PulsarURL       string
-	PulsarTenant    string
-	PulsarNamespace string
-
 	// Kubernetes config
 	KubernetesNamespace string
 
@@ -106,32 +99,12 @@ func NewWithGrpcProvider(config Config, provider grpcutils.GrpcProvider, db *gor
 	}
 
 	var notifier notification.Notifier
-	var client pulsar.Client
-	var producer pulsar.Producer
 	if config.NotifierProvider == "memory" {
 		log.Info("Using memory notifier")
 		notifier = notification.NewMemoryNotifier()
-	} else if config.NotifierProvider == "pulsar" {
-		log.Info("Using pulsar notifier")
-		pulsarNotifier, pulsarClient, pulsarProducer, err := createPulsarNotifer(config.PulsarURL, config.NotificationTopic)
-		notifier = pulsarNotifier
-		client = pulsarClient
-		producer = pulsarProducer
-		if err != nil {
-			log.Error("Failed to create pulsar notifier", zap.Error(err))
-			return nil, err
-		}
 	} else {
-		return nil, errors.New("invalid notifier provider, only memory and pulsar are supported")
+		return nil, errors.New("invalid notifier provider, only memory are supported")
 	}
-
-	if client != nil {
-		defer client.Close()
-	}
-	if producer != nil {
-		defer producer.Close()
-	}
-
 	coordinator, err := coordinator.NewCoordinator(ctx, db, notificationStore, notifier)
 	if err != nil {
 		return nil, err
@@ -189,27 +162,6 @@ func createMemberlistManager(namespace string, memberlistName string, podLabel s
 	return memberlist_manager, nil
 }
 
-func createPulsarNotifer(pulsarURL string, notificationTopic string) (*notification.PulsarNotifier, pulsar.Client, pulsar.Producer, error) {
-	client, err := pulsar.NewClient(pulsar.ClientOptions{
-		URL: pulsarURL,
-	})
-	if err != nil {
-		log.Error("Failed to create pulsar client", zap.Error(err))
-		return nil, nil, nil, err
-	}
-
-	producer, err := client.CreateProducer(pulsar.ProducerOptions{
-		Topic: notificationTopic,
-	})
-	if err != nil {
-		log.Error("Failed to create producer", zap.Error(err))
-		return nil, nil, nil, err
-	}
-
-	notifier := notification.NewPulsarNotifier(producer)
-	return notifier, client, producer, nil
-}
-
 func (s *Server) Close() error {
 	s.healthServer.Shutdown()
 	s.coordinator.Stop()

k8s/distributed-chroma/values.yaml

@@ -39,9 +39,6 @@ sysdb:
   replicaCount: 1
   env:
   flags:
-    pulsar-admin-url: "http://pulsar.chroma:8080"
-    pulsar-url: "pulsar://pulsar.chroma:6650"
-    notifier-provider: "pulsar"
 
 logService:
   image:
@@ -76,4 +73,3 @@ logServiceMigration:
     repository: 'local'
     tag: 'logservice-migration'
   databaseUrl: 'postgres://chroma:chroma@postgres:5432/log?sslmode=disable'
-