diff --git a/weed/mq/kafka/gateway/test_mock_handler.go b/weed/mq/kafka/gateway/test_mock_handler.go index ef0a012ef..25af11946 100644 --- a/weed/mq/kafka/gateway/test_mock_handler.go +++ b/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, diff --git a/weed/mq/kafka/integration/seaweedmq_handler_topics.go b/weed/mq/kafka/integration/seaweedmq_handler_topics.go index b2071fd00..2ca47d934 100644 --- a/weed/mq/kafka/integration/seaweedmq_handler_topics.go +++ b/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 diff --git a/weed/mq/kafka/protocol/handler.go b/weed/mq/kafka/protocol/handler.go index 0e726fd1d..c1225c5f4 100644 --- a/weed/mq/kafka/protocol/handler.go +++ b/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 { diff --git a/weed/mq/kafka/protocol/produce.go b/weed/mq/kafka/protocol/produce.go index 8a8453ea8..431ed1a40 100644 --- a/weed/mq/kafka/protocol/produce.go +++ b/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 diff --git a/weed/mq/kafka/protocol/topic_autocreate_test.go b/weed/mq/kafka/protocol/topic_autocreate_test.go new file mode 100644 index 000000000..9c4a6d81d --- /dev/null +++ b/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") + } +}