organize error

Author: mmreza79

errors.go

@@ -15,6 +15,12 @@ var (
 	errBuildProducer = "error creating Kafka producer: %v"
 	errApplyOptions  = "error applying options: %v"
 	errTopicNotFound = errors.New("requested topic was not found")
+	errTopicNoCreate = func(topic string) error {
+		return fmt.Errorf("topic '%s' does not exist but the manager is configured with NoCreate, so it will not attempt to create it", topic)
+	}
+	errTopicChecking = func(topic string, err error) error {
+		return fmt.Errorf("error checking topic '%s': %w", topic, err)
+	}
 )
 
 // this regex matches the package name + some hash info, if we're in gomod but not subpackages

topic_manager.go

@@ -172,17 +172,15 @@ func (m *topicManager) ensureExists(topic string, npar, rfactor int, config map[
 
 	partitions, err := m.Partitions(topic)
 
-	if err != nil {
-		if err != errTopicNotFound {
-			return fmt.Errorf("error checking topic: %v", err)
-		}
+	if err != nil && err != errTopicNotFound {
+		return errTopicChecking(topic, err)
 	}
 	// no topic yet, let's create it
 	if len(partitions) == 0 {
 
 		// (or not)
 		if m.topicManagerConfig.NoCreate {
-			return fmt.Errorf("topic `%s` does not exist but the manager is configured with NoCreate, so it will not attempt to create it", topic)
+			return errTopicNoCreate(topic)
 		}
 
 		return m.createTopic(topic,
@@ -195,7 +193,7 @@ func (m *topicManager) ensureExists(topic string, npar, rfactor int, config map[
 
 	// partitions do not match
 	if len(partitions) != npar {
-		return m.handleConfigMismatch(fmt.Sprintf("partition count mismatch for topic %s. Need %d, but existing topic has %d", topic, npar, len(partitions)))
+		return m.handleConfigMismatch(fmt.Sprintf("partition count mismatch for topic '%s'. Need %d, but existing topic has %d", topic, npar, len(partitions)))
 	}
 
 	// check additional config values via the cluster admin if our current version supports it
@@ -249,7 +247,7 @@ func (m *topicManager) waitForCreated(topic string) error {
 		case errTopicNotFound:
 			time.Sleep(time.Second)
 		default:
-			return fmt.Errorf("error checking topic: %w", err)
+			return errTopicChecking(topic, err)
 		}
 	}
 	return fmt.Errorf("waiting for topic %s to be created timed out", topic)

topic_manager_test.go

@@ -269,7 +269,7 @@ func TestTM_EnsureStreamExists(t *testing.T) {
 		)
 
 		err := tm.EnsureStreamExists(topic, npar)
-		require.Equal(t, err.Error(), "topic `some-topic` does not exist but the manager is configured with NoCreate, so it will not attempt to create it")
+		require.Equal(t, err, errTopicNoCreate(topic))
 	})
 	t.Run("fail", func(t *testing.T) {
 		tm, bm, ctrl := createTopicManager(t)