Accept empty Kafka config when visibility queue is set to internal (#…

…1238)
temporalio · Jan 28, 2021 · 485ae99 · 485ae99
1 parent e12c7a1
commit 485ae99
Showing 1 changed file with 5 additions and 1 deletion.
diff --git a/temporal/server.go b/temporal/server.go
@@ -290,8 +290,12 @@ func (s *Server) getServiceParams(
 		dynamicconfig.AdvancedVisibilityWritingMode,
 		common.GetDefaultAdvancedVisibilityWritingMode(s.so.config.Persistence.IsAdvancedVisibilityConfigExist()),
 	)()
+	visibilityQueue := dc.GetStringProperty(
+		dynamicconfig.VisibilityQueue,
+		common.VisibilityQueueInternal,
+	)()
 	isAdvancedVisEnabled := advancedVisMode != common.AdvancedVisibilityWritingModeOff
-	if isAdvancedVisEnabled {
+	if isAdvancedVisEnabled && (visibilityQueue == common.VisibilityQueueKafka || visibilityQueue == common.VisibilityQueueInternalWithDualProcessor) {
 		params.MessagingClient = messaging.NewKafkaClient(&s.so.config.Kafka, metricsClient, zap.NewNop(), s.logger, metricsScope, false, isAdvancedVisEnabled)
 	} else {
 		params.MessagingClient = nil