mq(kafka): don't drop an existing topic when auto-create races (#9998)

TopicExists can return a transient false-negative for a topic that is in
fact present (a broker/filer blip under load, or a just-created topic whose
existence cache is still stale). The metadata and produce handlers then tried
to auto-create, hit "topic already exists", treated that as a failure, and
dropped the topic from the Metadata response - so the client saw a spurious
UNKNOWN_TOPIC_OR_PARTITION right after the topic was created.

Return a sentinel ErrTopicAlreadyExists from the create paths and add
ensureTopicExists, which treats it as confirmation the topic exists. It also
folds in the repeated TopicExists -> invalidate -> recheck -> create logic
shared by every metadata handler (v0-v8) and both produce paths. v2+ produce
now auto-creates too, matching the auto.create.topics.enable=true behavior the
rest of the gateway already simulates.

weed/mq/kafka/gateway/test_mock_handler.go

@@ -62,7 +62,7 @@ func (m *mockSeaweedMQHandler) CreateTopic(topic string, partitions int32) error
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	if _, exists := m.topics[topic]; exists {
-		return fmt.Errorf("topic already exists")
+		return fmt.Errorf("%s: %w", topic, integration.ErrTopicAlreadyExists)
 	}
 	m.topics[topic] = &integration.KafkaTopicInfo{
 		Name:       topic,
@@ -75,7 +75,7 @@ func (m *mockSeaweedMQHandler) CreateTopicWithSchemas(name string, partitions in
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	if _, exists := m.topics[name]; exists {
-		return fmt.Errorf("topic already exists")
+		return fmt.Errorf("%s: %w", name, integration.ErrTopicAlreadyExists)
 	}
 	m.topics[name] = &integration.KafkaTopicInfo{
 		Name:       name,

weed/mq/kafka/integration/seaweedmq_handler_topics.go

@@ -2,6 +2,7 @@ package integration
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
@@ -15,6 +16,11 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
+// ErrTopicAlreadyExists is returned by the CreateTopic* methods when the topic
+// already exists in the filer. Callers performing auto-creation can treat it as
+// confirmation the topic exists rather than a hard failure.
+var ErrTopicAlreadyExists = errors.New("topic already exists")
+
 // CreateTopic creates a new topic in both Kafka registry and SeaweedMQ
 func (h *SeaweedMQHandler) CreateTopic(name string, partitions int32) error {
 	return h.CreateTopicWithSchema(name, partitions, nil)
@@ -29,7 +35,7 @@ func (h *SeaweedMQHandler) CreateTopicWithSchema(name string, partitions int32,
 func (h *SeaweedMQHandler) CreateTopicWithSchemas(name string, partitions int32, keyRecordType *schema_pb.RecordType, valueRecordType *schema_pb.RecordType) error {
 	// Check if topic already exists in filer
 	if h.checkTopicInFiler(name) {
-		return fmt.Errorf("topic %s already exists", name)
+		return fmt.Errorf("%s: %w", name, ErrTopicAlreadyExists)
 	}
 
 	// Create SeaweedMQ topic reference
@@ -90,7 +96,7 @@ func (h *SeaweedMQHandler) CreateTopicWithSchemas(name string, partitions int32,
 func (h *SeaweedMQHandler) CreateTopicWithRecordType(name string, partitions int32, flatSchema *schema_pb.RecordType, keyColumns []string) error {
 	// Check if topic already exists in filer
 	if h.checkTopicInFiler(name) {
-		return fmt.Errorf("topic %s already exists", name)
+		return fmt.Errorf("%s: %w", name, ErrTopicAlreadyExists)
 	}
 
 	// Create SeaweedMQ topic reference

weed/mq/kafka/protocol/handler.go

@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/binary"
+	"errors"
 	"fmt"
 	"hash/fnv"
 	"io"
@@ -1317,28 +1318,8 @@ func (h *Handler) HandleMetadataV0(correlationID uint32, requestBody []byte) ([]
 		topicsToReturn = h.seaweedMQHandler.ListTopics()
 	} else {
 		for _, name := range requestedTopics {
-			if h.seaweedMQHandler.TopicExists(name) {
+			if h.ensureTopicExists(name, h.GetDefaultPartitions()) {
 				topicsToReturn = append(topicsToReturn, name)
-			} else {
-				// Topic doesn't exist according to current cache, check broker directly
-				// This handles the race condition where producers just created topics
-				// and consumers are requesting metadata before cache TTL expires
-				glog.V(3).Infof("[METADATA v0] Topic %s not in cache, checking broker directly", name)
-				h.seaweedMQHandler.InvalidateTopicExistsCache(name)
-				if h.seaweedMQHandler.TopicExists(name) {
-					glog.V(3).Infof("[METADATA v0] Topic %s found on broker after cache refresh", name)
-					topicsToReturn = append(topicsToReturn, name)
-				} else {
-					glog.V(3).Infof("[METADATA v0] Topic %s not found, auto-creating with default partitions", name)
-					// Auto-create topic (matches Kafka's auto.create.topics.enable=true)
-					if err := h.createTopicWithSchemaSupport(name, h.GetDefaultPartitions()); err != nil {
-						glog.V(2).Infof("[METADATA v0] Failed to auto-create topic %s: %v", name, err)
-						// Don't add to topicsToReturn - client will get error
-					} else {
-						glog.V(2).Infof("[METADATA v0] Successfully auto-created topic %s", name)
-						topicsToReturn = append(topicsToReturn, name)
-					}
-				}
 			}
 		}
 	}
@@ -1412,24 +1393,8 @@ func (h *Handler) HandleMetadataV1(correlationID uint32, requestBody []byte) ([]
 		topicsToReturn = h.seaweedMQHandler.ListTopics()
 	} else {
 		for _, name := range requestedTopics {
-			if h.seaweedMQHandler.TopicExists(name) {
+			if h.ensureTopicExists(name, h.GetDefaultPartitions()) {
 				topicsToReturn = append(topicsToReturn, name)
-			} else {
-				// Topic doesn't exist according to current cache, check broker directly
-				glog.V(3).Infof("[METADATA v1] Topic %s not in cache, checking broker directly", name)
-				h.seaweedMQHandler.InvalidateTopicExistsCache(name)
-				if h.seaweedMQHandler.TopicExists(name) {
-					glog.V(3).Infof("[METADATA v1] Topic %s found on broker after cache refresh", name)
-					topicsToReturn = append(topicsToReturn, name)
-				} else {
-					glog.V(3).Infof("[METADATA v1] Topic %s not found, auto-creating with default partitions", name)
-					if err := h.createTopicWithSchemaSupport(name, h.GetDefaultPartitions()); err != nil {
-						glog.V(2).Infof("[METADATA v1] Failed to auto-create topic %s: %v", name, err)
-					} else {
-						glog.V(2).Infof("[METADATA v1] Successfully auto-created topic %s", name)
-						topicsToReturn = append(topicsToReturn, name)
-					}
-				}
 			}
 		}
 	}
@@ -1546,24 +1511,8 @@ func (h *Handler) HandleMetadataV2(correlationID uint32, requestBody []byte) ([]
 		topicsToReturn = h.seaweedMQHandler.ListTopics()
 	} else {
 		for _, name := range requestedTopics {
-			if h.seaweedMQHandler.TopicExists(name) {
+			if h.ensureTopicExists(name, h.GetDefaultPartitions()) {
 				topicsToReturn = append(topicsToReturn, name)
-			} else {
-				// Topic doesn't exist according to current cache, check broker directly
-				glog.V(3).Infof("[METADATA v2] Topic %s not in cache, checking broker directly", name)
-				h.seaweedMQHandler.InvalidateTopicExistsCache(name)
-				if h.seaweedMQHandler.TopicExists(name) {
-					glog.V(3).Infof("[METADATA v2] Topic %s found on broker after cache refresh", name)
-					topicsToReturn = append(topicsToReturn, name)
-				} else {
-					glog.V(3).Infof("[METADATA v2] Topic %s not found, auto-creating with default partitions", name)
-					if err := h.createTopicWithSchemaSupport(name, h.GetDefaultPartitions()); err != nil {
-						glog.V(2).Infof("[METADATA v2] Failed to auto-create topic %s: %v", name, err)
-					} else {
-						glog.V(2).Infof("[METADATA v2] Successfully auto-created topic %s", name)
-						topicsToReturn = append(topicsToReturn, name)
-					}
-				}
 			}
 		}
 	}
@@ -1670,24 +1619,8 @@ func (h *Handler) HandleMetadataV3V4(correlationID uint32, requestBody []byte) (
 		topicsToReturn = h.seaweedMQHandler.ListTopics()
 	} else {
 		for _, name := range requestedTopics {
-			if h.seaweedMQHandler.TopicExists(name) {
+			if h.ensureTopicExists(name, h.GetDefaultPartitions()) {
 				topicsToReturn = append(topicsToReturn, name)
-			} else {
-				// Topic doesn't exist according to current cache, check broker directly
-				glog.V(3).Infof("[METADATA v3/v4] Topic %s not in cache, checking broker directly", name)
-				h.seaweedMQHandler.InvalidateTopicExistsCache(name)
-				if h.seaweedMQHandler.TopicExists(name) {
-					glog.V(3).Infof("[METADATA v3/v4] Topic %s found on broker after cache refresh", name)
-					topicsToReturn = append(topicsToReturn, name)
-				} else {
-					glog.V(3).Infof("[METADATA v3/v4] Topic %s not found, auto-creating with default partitions", name)
-					if err := h.createTopicWithSchemaSupport(name, h.GetDefaultPartitions()); err != nil {
-						glog.V(2).Infof("[METADATA v3/v4] Failed to auto-create topic %s: %v", name, err)
-					} else {
-						glog.V(2).Infof("[METADATA v3/v4] Successfully auto-created topic %s", name)
-						topicsToReturn = append(topicsToReturn, name)
-					}
-				}
 			}
 		}
 	}
@@ -1828,44 +1761,15 @@ func (h *Handler) handleMetadataV5ToV8(correlationID uint32, requestBody []byte,
 		// 4. Client will call CreateTopics for non-existent topics
 		// 5. Then request metadata again to see the created topics
 		for _, topic := range requestedTopics {
+			partitions := h.GetDefaultPartitions()
 			if isSystemTopic(topic) {
-				// Always try to auto-create system topics during metadata requests
-				glog.V(3).Infof("[METADATA v%d] Ensuring system topic %s exists during metadata request", apiVersion, topic)
-				if !h.seaweedMQHandler.TopicExists(topic) {
-					glog.V(3).Infof("[METADATA v%d] Auto-creating system topic %s during metadata request", apiVersion, topic)
-					if err := h.createTopicWithSchemaSupport(topic, 1); err != nil {
-						glog.V(0).Infof("[METADATA v%d] Failed to auto-create system topic %s: %v", apiVersion, topic, err)
-						// Continue without adding to topicsToReturn - client will get UNKNOWN_TOPIC_OR_PARTITION
-					} else {
-						glog.V(3).Infof("[METADATA v%d] Successfully auto-created system topic %s", apiVersion, topic)
-					}
-				} else {
-					glog.V(3).Infof("[METADATA v%d] System topic %s already exists", apiVersion, topic)
-				}
+				// System topics use a single partition. ensureTopicExists tolerates
+				// the auto-create race, so a false result means the topic genuinely
+				// does not exist - don't advertise it with error_code=0.
+				partitions = 1
+			}
+			if h.ensureTopicExists(topic, partitions) {
 				topicsToReturn = append(topicsToReturn, topic)
-			} else if h.seaweedMQHandler.TopicExists(topic) {
-				topicsToReturn = append(topicsToReturn, topic)
-			} else {
-				// Topic doesn't exist according to current cache, but let's check broker directly
-				// This handles the race condition where producers just created topics
-				// and consumers are requesting metadata before cache TTL expires
-				glog.V(3).Infof("[METADATA v%d] Topic %s not in cache, checking broker directly", apiVersion, topic)
-				// Force cache invalidation to do fresh broker check
-				h.seaweedMQHandler.InvalidateTopicExistsCache(topic)
-				if h.seaweedMQHandler.TopicExists(topic) {
-					glog.V(3).Infof("[METADATA v%d] Topic %s found on broker after cache refresh", apiVersion, topic)
-					topicsToReturn = append(topicsToReturn, topic)
-				} else {
-					glog.V(3).Infof("[METADATA v%d] Topic %s not found on broker, auto-creating with default partitions", apiVersion, topic)
-					// Auto-create non-system topics with default partitions (matches Kafka behavior)
-					if err := h.createTopicWithSchemaSupport(topic, h.GetDefaultPartitions()); err != nil {
-						glog.V(2).Infof("[METADATA v%d] Failed to auto-create topic %s: %v", apiVersion, topic, err)
-						// Don't add to topicsToReturn - client will get UNKNOWN_TOPIC_OR_PARTITION
-					} else {
-						glog.V(2).Infof("[METADATA v%d] Successfully auto-created topic %s", apiVersion, topic)
-						topicsToReturn = append(topicsToReturn, topic)
-					}
-				}
 			}
 		}
 		glog.V(3).Infof("[METADATA v%d] Returning topics: %v (requested: %v)", apiVersion, topicsToReturn, requestedTopics)
@@ -4067,6 +3971,39 @@ func (h *Handler) handleInitProducerId(correlationID uint32, apiVersion uint16,
 	return response, nil
 }
 
+// ensureTopicExists reports whether the named topic exists, auto-creating it
+// when missing (matching auto.create.topics.enable=true).
+//
+// TopicExists can return a transient false negative for a topic that is in fact
+// present: a broker/filer blip, or a just-created topic whose existence cache is
+// still stale. When that happens the auto-create attempt fails with
+// ErrTopicAlreadyExists - which confirms the topic exists. Treating that as
+// success (rather than dropping the topic) is what keeps producers from getting
+// a spurious UNKNOWN_TOPIC_OR_PARTITION right after the topic was created.
+func (h *Handler) ensureTopicExists(topicName string, partitions int32) bool {
+	if h.seaweedMQHandler.TopicExists(topicName) {
+		return true
+	}
+	// Existence cache may be stale right after creation; force a fresh broker check.
+	h.seaweedMQHandler.InvalidateTopicExistsCache(topicName)
+	if h.seaweedMQHandler.TopicExists(topicName) {
+		return true
+	}
+	if err := h.createTopicWithSchemaSupport(topicName, partitions); err != nil {
+		if errors.Is(err, integration.ErrTopicAlreadyExists) {
+			// The topic exists; the recheck above just cached a stale negative.
+			// Clear it so direct TopicExists callers don't see the false negative.
+			h.seaweedMQHandler.InvalidateTopicExistsCache(topicName)
+			return true
+		}
+		glog.V(2).Infof("[METADATA] Failed to auto-create topic %s: %v", topicName, err)
+		return false
+	}
+	// Topic just created - clear the negative cache so callers see it immediately.
+	h.seaweedMQHandler.InvalidateTopicExistsCache(topicName)
+	return true
+}
+
 // createTopicWithSchemaSupport creates a topic with optional schema integration
 // This function creates topics with schema support when schema management is enabled
 func (h *Handler) createTopicWithSchemaSupport(topicName string, partitions int32) error {

weed/mq/kafka/protocol/produce.go

@@ -7,7 +7,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/mq/kafka/compression"
 	"github.com/seaweedfs/seaweedfs/weed/mq/kafka/schema"
 	"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
@@ -78,24 +77,7 @@ func (h *Handler) handleProduceV0V1(ctx context.Context, correlationID uint32, a
 		offset += 4
 
 		// Check if topic exists, auto-create if it doesn't (simulates auto.create.topics.enable=true)
-		topicExists := h.seaweedMQHandler.TopicExists(topicName)
-
-		_ = h.seaweedMQHandler.ListTopics() // existingTopics
-		if !topicExists {
-			// Use schema-aware topic creation for auto-created topics with configurable default partitions
-			defaultPartitions := h.GetDefaultPartitions()
-			glog.V(1).Infof("[PRODUCE] Topic %s does not exist, auto-creating with %d partitions", topicName, defaultPartitions)
-			if err := h.createTopicWithSchemaSupport(topicName, defaultPartitions); err != nil {
-				glog.V(0).Infof("[PRODUCE] ERROR: Failed to auto-create topic %s: %v", topicName, err)
-			} else {
-				glog.V(1).Infof("[PRODUCE] Successfully auto-created topic %s", topicName)
-				// Invalidate cache immediately after creation so consumers can find it
-				h.seaweedMQHandler.InvalidateTopicExistsCache(topicName)
-				topicExists = true
-			}
-		} else {
-			glog.V(2).Infof("[PRODUCE] Topic %s already exists", topicName)
-		}
+		topicExists := h.ensureTopicExists(topicName, h.GetDefaultPartitions())
 
 		// Response: topic_name_size(2) + topic_name + partitions_array
 		response = append(response, byte(topicNameSize>>8), byte(topicNameSize))
@@ -701,8 +683,11 @@ func (h *Handler) handleProduceV2Plus(ctx context.Context, correlationID uint32,
 			var baseOffset int64 = 0
 			currentTime := time.Now().UnixNano()
 
-			// Check if topic exists; for v2+ do NOT auto-create
-			topicExists := h.seaweedMQHandler.TopicExists(topicName)
+			// Check if topic exists, auto-creating if needed. ensureTopicExists
+			// tolerates a transient TopicExists false-negative for a topic that
+			// already exists, which otherwise surfaces as a spurious
+			// UNKNOWN_TOPIC_OR_PARTITION right after the topic was created.
+			topicExists := h.ensureTopicExists(topicName, h.GetDefaultPartitions())
 
 			if !topicExists {
 				errorCode = 3 // UNKNOWN_TOPIC_OR_PARTITION

weed/mq/kafka/protocol/topic_autocreate_test.go

@@ -0,0 +1,74 @@
+package protocol
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/mq/kafka/integration"
+)
+
+// autoCreateMockHandler models the flaky-existence scenario: TopicExists reports
+// the topic missing (a transient broker/filer false-negative), while a create
+// attempt reports the topic already exists. It reuses FastMockHandler for the
+// rest of the interface and overrides only the two methods under test.
+type autoCreateMockHandler struct {
+	*FastMockHandler
+	exists    bool
+	createErr error
+}
+
+func (h *autoCreateMockHandler) TopicExists(name string) bool { return h.exists }
+
+func (h *autoCreateMockHandler) CreateTopic(name string, partitions int32) error {
+	return h.createErr
+}
+
+// TestEnsureTopicExistsTreatsAlreadyExistsAsSuccess reproduces the offset-resumption
+// flake: TopicExists returns a false-negative for a topic that is in fact present,
+// so the auto-create attempt fails with ErrTopicAlreadyExists. The topic must still
+// be reported as existing, otherwise the client sees UNKNOWN_TOPIC_OR_PARTITION.
+func TestEnsureTopicExistsTreatsAlreadyExistsAsSuccess(t *testing.T) {
+	handler := &Handler{
+		seaweedMQHandler: &autoCreateMockHandler{
+			FastMockHandler: &FastMockHandler{},
+			exists:          false,
+			createErr:       fmt.Errorf("%s: %w", "offset-resumption", integration.ErrTopicAlreadyExists),
+		},
+	}
+
+	if !handler.ensureTopicExists("offset-resumption", 1) {
+		t.Fatal("ensureTopicExists must report true when the create attempt finds the topic already exists")
+	}
+}
+
+// TestEnsureTopicExistsReportsGenuineFailure confirms a real create failure (not
+// "already exists") still results in the topic being treated as absent.
+func TestEnsureTopicExistsReportsGenuineFailure(t *testing.T) {
+	handler := &Handler{
+		seaweedMQHandler: &autoCreateMockHandler{
+			FastMockHandler: &FastMockHandler{},
+			exists:          false,
+			createErr:       fmt.Errorf("broker unavailable"),
+		},
+	}
+
+	if handler.ensureTopicExists("missing-topic", 1) {
+		t.Fatal("ensureTopicExists must report false when the topic is absent and creation genuinely fails")
+	}
+}
+
+// TestEnsureTopicExistsWhenPresent covers the common path where the topic is
+// already known to exist and no create attempt is needed.
+func TestEnsureTopicExistsWhenPresent(t *testing.T) {
+	handler := &Handler{
+		seaweedMQHandler: &autoCreateMockHandler{
+			FastMockHandler: &FastMockHandler{},
+			exists:          true,
+			createErr:       fmt.Errorf("CreateTopic should not be called when the topic exists"),
+		},
+	}
+
+	if !handler.ensureTopicExists("present-topic", 1) {
+		t.Fatal("ensureTopicExists must report true for an existing topic")
+	}
+}