ensure topic exists as more flexible topic creation option

kafka/mock/kazoo.go

@@ -49,6 +49,16 @@ func (_mr *_MockTopicManagerRecorder) EnsureStreamExists(arg0, arg1 interface{})
 	return _mr.mock.ctrl.RecordCall(_mr.mock, "EnsureStreamExists", arg0, arg1)
 }
 
+func (_m *MockTopicManager) EnsureTopicExists(topic string, npar int, rfactor int, config map[string]string) error {
+	ret := _m.ctrl.Call(_m, "EnsureTopicExists", topic, npar, rfactor, config)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+func (_mr *_MockTopicManagerRecorder) EnsureTopicExists(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
+	return _mr.mock.ctrl.RecordCall(_mr.mock, "EnsureTopicExists", arg0, arg1, arg2, arg3)
+}
+
 func (_m *MockTopicManager) Partitions(topic string) ([]int32, error) {
 	ret := _m.ctrl.Call(_m, "Partitions", topic)
 	ret0, _ := ret[0].([]int32)

kafka/topic_manager.go

@@ -16,6 +16,9 @@ type TopicManager interface {
 	EnsureTableExists(topic string, npar int) error
 	// EnsureStreamExists checks that a stream topic exists, or create one if possible
 	EnsureStreamExists(topic string, npar int) error
+	// EnsureTopicExists checks that a topic exists, or create one if possible,
+	// enforcing the given configuration
+	EnsureTopicExists(topic string, npar, rfactor int, config map[string]string) error
 
 	// Partitions returns the number of partitions of a topic, that are assigned to the running
 	// instance, i.e. it doesn't represent all partitions of a topic.
@@ -66,6 +69,10 @@ func (m *saramaTopicManager) EnsureTableExists(topic string, npar int) error {
 	return nil
 }
 
+func (m *saramaTopicManager) EnsureTopicExists(topic string, npar, rfactor int, config map[string]string) error {
+	return fmt.Errorf("not implemented in SaramaTopicManager")
+}
+
 // TopicManagerConfig contains the configuration to access the Zookeeper servers
 // as well as the desired options of to create tables and stream topics.
 type TopicManagerConfig struct {
@@ -149,6 +156,13 @@ func (m *topicManager) EnsureStreamExists(topic string, npar int) error {
 	return m.checkPartitions(topic, npar)
 }
 
+func (m *topicManager) EnsureTopicExists(topic string, npar, rfactor int, config map[string]string) error {
+	if err := checkTopic(m.zk, topic, npar, rfactor, config); err != nil {
+		return err
+	}
+	return m.checkPartitions(topic, npar)
+}
+
 func (m *topicManager) Partitions(topic string) ([]int32, error) {
 	tl, err := m.zk.Topics()
 	if err != nil {
@@ -176,14 +190,22 @@ func checkTopic(kz kzoo, topic string, npar int, rfactor int, cfg map[string]str
 	if err != nil {
 		return err
 	}
-	if ok {
-		return nil
+	if !ok {
+		err = kz.CreateTopic(topic, npar, rfactor, cfg)
+		if err != nil {
+			return err
+		}
 	}
-	err = kz.CreateTopic(topic, npar, rfactor, cfg)
+	// topic exists, check if config the same
+	c, err := kz.Topic(topic).Config()
 	if err != nil {
 		return err
 	}
-
+	for k, v := range cfg {
+		if c[k] != v {
+			return fmt.Errorf("expected %s=%s, but found %s", k, cfg[k], c[k])
+		}
+	}
 	return nil
 }
 

mock/kafka.go

@@ -148,6 +148,16 @@ func (_mr *_MockTopicManagerRecorder) EnsureTableExists(arg0, arg1 interface{})
 	return _mr.mock.ctrl.RecordCall(_mr.mock, "EnsureTableExists", arg0, arg1)
 }
 
+func (_m *MockTopicManager) EnsureTopicExists(_param0 string, _param1 int, _param2 int, _param3 map[string]string) error {
+	ret := _m.ctrl.Call(_m, "EnsureTopicExists", _param0, _param1, _param2, _param3)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+func (_mr *_MockTopicManagerRecorder) EnsureTopicExists(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
+	return _mr.mock.ctrl.RecordCall(_mr.mock, "EnsureTopicExists", arg0, arg1, arg2, arg3)
+}
+
 func (_m *MockTopicManager) Partitions(_param0 string) ([]int32, error) {
 	ret := _m.ctrl.Call(_m, "Partitions", _param0)
 	ret0, _ := ret[0].([]int32)

tester/tester.go

@@ -396,6 +396,11 @@ func (tm *topicMgrMock) EnsureStreamExists(topic string, npar int) error {
 	return nil
 }
 
+// EnsureTopicExists checks that a stream exists, or create one if possible
+func (tm *topicMgrMock) EnsureTopicExists(topic string, npar, rfactor int, config map[string]string) error {
+	return nil
+}
+
 // Partitions returns the number of partitions of a topic, that are assigned to the running
 // instance, i.e. it doesn't represent all partitions of a topic.
 func (tm *topicMgrMock) Partitions(topic string) ([]int32, error) {