Prefix-defined Kafka topics (#1254)

* Prefix-defined Kafka topics

* Fix current state server test

5 files changed, 2 insertions, 17 deletions

diff --git a/cmd/dendrite-demo-libp2p/main.go b/cmd/dendrite-demo-libp2p/main.go
index 8c28014a..e3035547 100644
--- a/cmd/dendrite-demo-libp2p/main.go
+++ b/cmd/dendrite-demo-libp2p/main.go
@@ -118,11 +118,6 @@ func main() {
 	cfg.Global.PrivateKey = privKey
 	cfg.Global.KeyID = gomatrixserverlib.KeyID(fmt.Sprintf("ed25519:%s", *instanceName))
 	cfg.Global.Kafka.UseNaffka = true
-	cfg.Global.Kafka.Topics.OutputRoomEvent = "roomserverOutput"
-	cfg.Global.Kafka.Topics.OutputClientData = "clientapiOutput"
-	cfg.Global.Kafka.Topics.OutputTypingEvent = "typingServerOutput"
-	cfg.Global.Kafka.Topics.OutputSendToDeviceEvent = "sendToDeviceOutput"
-	cfg.Global.Kafka.Topics.OutputKeyChangeEvent = "keyChangeOutput"
 	cfg.FederationSender.FederationMaxRetries = 6
 	cfg.UserAPI.AccountDatabase.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-account.db", *instanceName))
 	cfg.UserAPI.DeviceDatabase.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-device.db", *instanceName))
diff --git a/cmd/dendrite-demo-yggdrasil/main.go b/cmd/dendrite-demo-yggdrasil/main.go
index cf4c7735..33ed1542 100644
--- a/cmd/dendrite-demo-yggdrasil/main.go
+++ b/cmd/dendrite-demo-yggdrasil/main.go
@@ -73,11 +73,6 @@ func main() {
 	cfg.Global.PrivateKey = ygg.SigningPrivateKey()
 	cfg.Global.KeyID = gomatrixserverlib.KeyID(signing.KeyID)
 	cfg.Global.Kafka.UseNaffka = true
-	cfg.Global.Kafka.Topics.OutputRoomEvent = "roomserverOutput"
-	cfg.Global.Kafka.Topics.OutputClientData = "clientapiOutput"
-	cfg.Global.Kafka.Topics.OutputTypingEvent = "typingServerOutput"
-	cfg.Global.Kafka.Topics.OutputSendToDeviceEvent = "sendToDeviceOutput"
-	cfg.Global.Kafka.Topics.OutputKeyChangeEvent = "keyChangeOutput"
 	cfg.FederationSender.FederationMaxRetries = 8
 	cfg.UserAPI.AccountDatabase.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-account.db", *instanceName))
 	cfg.UserAPI.DeviceDatabase.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-device.db", *instanceName))
diff --git a/cmd/dendritejs/main.go b/cmd/dendritejs/main.go
index ce7812fa..f350a938 100644
--- a/cmd/dendritejs/main.go
+++ b/cmd/dendritejs/main.go
@@ -174,10 +174,6 @@ func main() {
 	cfg.KeyServer.Database.ConnectionString = "file:/idb/dendritejs_e2ekey.db"
 	cfg.Global.Kafka.UseNaffka = true
 	cfg.Global.Kafka.Database.ConnectionString = "file:/idb/dendritejs_naffka.db"
-	cfg.Global.Kafka.Topics.OutputTypingEvent = "output_typing_event"
-	cfg.Global.Kafka.Topics.OutputSendToDeviceEvent = "output_send_to_device_event"
-	cfg.Global.Kafka.Topics.OutputClientData = "output_client_data"
-	cfg.Global.Kafka.Topics.OutputRoomEvent = "output_room_event"
 	cfg.Global.TrustedIDServers = []string{
 		"matrix.org", "vector.im",
 	}
diff --git a/cmd/roomserver-integration-tests/main.go b/cmd/roomserver-integration-tests/main.go
index 4d3095be..a9c88ae1 100644
--- a/cmd/roomserver-integration-tests/main.go
+++ b/cmd/roomserver-integration-tests/main.go
@@ -29,6 +29,7 @@ import (
 	"net/http"
 
 	"github.com/matrix-org/dendrite/internal/caching"
+	"github.com/matrix-org/dendrite/internal/config"
 	"github.com/matrix-org/dendrite/internal/test"
 	"github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/roomserver/inthttp"
@@ -240,7 +241,7 @@ func testRoomserver(input []string, wantOutput []string, checkQueries func(api.R
 		panic(err)
 	}
 
-	outputTopic := string(cfg.Global.Kafka.Topics.OutputRoomEvent)
+	outputTopic := cfg.Global.Kafka.TopicFor(config.TopicOutputRoomEvent)
 
 	err = exe.DeleteTopic(outputTopic)
 	if err != nil {
diff --git a/cmd/syncserver-integration-tests/main.go b/cmd/syncserver-integration-tests/main.go
index 76fa0eba..bba8fd96 100644
--- a/cmd/syncserver-integration-tests/main.go
+++ b/cmd/syncserver-integration-tests/main.go
@@ -134,8 +134,6 @@ func startSyncServer() (*exec.Cmd, chan error) {
 	// TODO use the address assigned by the config generator rather than clobbering.
 	cfg.Global.ServerName = "localhost"
 	cfg.SyncAPI.Listen = config.Address(syncserverAddr)
-	cfg.Global.Kafka.Topics.OutputRoomEvent = config.Topic(inputTopic)
-	cfg.Global.Kafka.Topics.OutputClientData = config.Topic(clientTopic)
 
 	if err := test.WriteConfig(cfg, dir); err != nil {
 		panic(err)