diff --git a/cmd/broker/main.go b/cmd/broker/main.go
index 44ff9bcd..0279254f 100644
--- a/cmd/broker/main.go
+++ b/cmd/broker/main.go
@@ -33,6 +33,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/twmb/franz-go/pkg/kerr"
 	"github.com/twmb/franz-go/pkg/kmsg"
 
 	"github.com/KafScale/platform/internal/s3env"
@@ -1040,8 +1041,57 @@ func (h *handler) acquirePartitionLeases(ctx context.Context, req *kmsg.ProduceR
 	}
 	results := h.leaseManager.AcquireAll(ctx, partitions)
 	var errs map[metadata.PartitionID]error
+	type topicLeaseMetadata struct {
+		partitions map[int32]struct{}
+		err        error
+	}
+	var checkedTopics map[string]topicLeaseMetadata
+	var refreshed bool
+	var refreshErr error
 	for _, r := range results {
 		if r.Err != nil {
+			if errors.Is(r.Err, metadata.ErrNotOwner) {
+				// A previous invalid request may have acquired a lease for a
+				// nonexistent partition. Check metadata before redirecting it.
+				// Refresh once to avoid rejecting a valid partition from an old
+				// snapshot. Healthy owned requests skip this work entirely.
+				if !refreshed {
+					refreshed = true
+					if store, ok := h.store.(interface{ RefreshSnapshot(context.Context) error }); ok {
+						refreshErr = store.RefreshSnapshot(ctx)
+					}
+				}
+				if checkedTopics == nil {
+					checkedTopics = make(map[string]topicLeaseMetadata)
+				}
+				view, checked := checkedTopics[r.Partition.Topic]
+				if !checked && refreshErr == nil {
+					if meta, err := h.store.Metadata(ctx, []string{r.Partition.Topic}); err == nil && meta != nil {
+						view.err = metadata.ErrUnknownTopic
+						for _, topic := range meta.Topics {
+							if topic.Topic == nil || *topic.Topic != r.Partition.Topic {
+								continue
+							}
+							view.err = kerr.ErrorForCode(topic.ErrorCode)
+							if view.err == nil {
+								view.partitions = make(map[int32]struct{})
+								for _, part := range topic.Partitions {
+									view.partitions[part.Partition] = struct{}{}
+								}
+							}
+							break
+						}
+					}
+					checkedTopics[r.Partition.Topic] = view
+				}
+				if view.err != nil {
+					r.Err = view.err
+				} else if view.partitions != nil {
+					if _, exists := view.partitions[r.Partition.Partition]; !exists {
+						r.Err = metadata.ErrUnknownTopic
+					}
+				}
+			}
 			if errs == nil {
 				errs = make(map[metadata.PartitionID]error)
 			}
@@ -1137,21 +1187,24 @@ func (h *handler) handleProduce(ctx context.Context, header *protocol.RequestHea
 				continue
 			}
 			if leaseErr, hasErr := leaseErrors[metadata.PartitionID{Topic: topic.Topic, Partition: part.Partition}]; hasErr {
-				if errors.Is(leaseErr, metadata.ErrNotOwner) || errors.Is(leaseErr, metadata.ErrShuttingDown) {
-					p := kmsg.NewProduceResponseTopicPartition()
-					p.Partition = part.Partition
-					p.ErrorCode = protocol.NOT_LEADER_OR_FOLLOWER
-					partitionResponses = append(partitionResponses, p)
-					if h.traceKafka {
-						h.logger.Debug("produce rejected: not partition owner", "topic", topic.Topic, "partition", part.Partition)
-					}
-					continue
-				}
 				p := kmsg.NewProduceResponseTopicPartition()
 				p.Partition = part.Partition
-				p.ErrorCode = protocol.REQUEST_TIMED_OUT
+				var topicErr *kerr.Error
+				switch {
+				case errors.Is(leaseErr, metadata.ErrUnknownTopic):
+					p.ErrorCode = protocol.UNKNOWN_TOPIC_OR_PARTITION
+				case errors.Is(leaseErr, metadata.ErrNotOwner), errors.Is(leaseErr, metadata.ErrShuttingDown):
+					p.ErrorCode = protocol.NOT_LEADER_OR_FOLLOWER
+				case errors.As(leaseErr, &topicErr):
+					p.ErrorCode = topicErr.Code
+				default:
+					p.ErrorCode = protocol.REQUEST_TIMED_OUT
+					h.logger.Warn("partition lease acquire failed", "topic", topic.Topic, "partition", part.Partition, "error", leaseErr)
+				}
 				partitionResponses = append(partitionResponses, p)
-				h.logger.Warn("partition lease acquire failed", "topic", topic.Topic, "partition", part.Partition, "error", leaseErr)
+				if h.traceKafka {
+					h.logger.Debug("produce rejected", "topic", topic.Topic, "partition", part.Partition, "error_code", p.ErrorCode)
+				}
 				continue
 			}
 			if h.s3Health.State() != broker.S3StateHealthy {
@@ -1182,6 +1235,8 @@ func (h *handler) handleProduce(ctx context.Context, header *protocol.RequestHea
 					p.ErrorCode = protocol.NOT_LEADER_OR_FOLLOWER
 				} else if errors.Is(err, metadata.ErrInvalidTopic) {
 					p.ErrorCode = protocol.INVALID_TOPIC_EXCEPTION
+				} else if errors.Is(err, metadata.ErrUnknownTopic) {
+					p.ErrorCode = protocol.UNKNOWN_TOPIC_OR_PARTITION
 				} else if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
 					p.ErrorCode = protocol.REQUEST_TIMED_OUT
 				} else if errors.Is(err, storage.ErrObjectStorage) {
@@ -2274,6 +2329,9 @@ func (h *handler) loadPartitionLog(ctx context.Context, topic string, partition
 				if err := h.ensureTopic(ctx, topic, partition); err != nil {
 					return nil, err
 				}
+				// An existing topic may still lack this partition. Retry its
+				// lookup once, including after a concurrent topic creation.
+				autoCreate = false
 				continue
 			}
 			return nil, err
diff --git a/cmd/broker/produce_lease_metadata_errors_test.go b/cmd/broker/produce_lease_metadata_errors_test.go
new file mode 100644
index 00000000..e1c82c02
--- /dev/null
+++ b/cmd/broker/produce_lease_metadata_errors_test.go
@@ -0,0 +1,94 @@
+// Copyright 2025 Alexander Alten (novatechflow), NovaTechflow (novatechflow.com).
+// This project is supported and financed by Scalytics, Inc. (www.scalytics.io).
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/KafScale/platform/pkg/metadata"
+	"github.com/KafScale/platform/pkg/protocol"
+	"github.com/KafScale/platform/pkg/storage"
+	"github.com/twmb/franz-go/pkg/kmsg"
+)
+
+func leaseMetadataErrorBrokers(t *testing.T) (*metadata.EtcdStore, *handler, *handler) {
+	t.Helper()
+	store := newEtcdTopicStore(t, context.Background())
+	a := newTestHandler(store)
+	b := newHandler(store, storage.NewMemoryS3Client(), protocol.MetadataBroker{NodeID: 2}, testLogger())
+	for _, h := range []*handler{a, b} {
+		t.Cleanup(h.coordinator.Stop)
+		t.Cleanup(h.leaseManager.ReleaseAll)
+		h.autoCreateTopics = true
+	}
+	return store, a, b
+}
+func TestProduceInvalidTopicNameAcrossLeaseOwners(t *testing.T) {
+	store, a, b := leaseMetadataErrorBrokers(t)
+	for _, name := range invalidTopicNames {
+		t.Run(name, func(t *testing.T) {
+			for _, h := range []*handler{a, b} {
+				raw := producePromptly(t, h, unknownPartitionProduce(name, 0, -1, 0))
+				p := decodeKmsgResponse(t, 3, raw, kmsg.NewPtrProduceResponse).Topics[0].Partitions[0]
+				if p.ErrorCode != protocol.INVALID_TOPIC_EXCEPTION {
+					t.Fatalf("broker %d: got %d, want INVALID_TOPIC_EXCEPTION", h.brokerInfo.NodeID, p.ErrorCode)
+				}
+			}
+			if topicExists(t, context.Background(), store, name) {
+				t.Fatal("invalid topic was created")
+			}
+		})
+	}
+	if len(a.logs) != 0 || len(b.logs) != 0 {
+		t.Fatal("invalid requests initialized partition logs")
+	}
+}
+
+type topicErrorLeaseStore struct {
+	*metadata.EtcdStore
+	code int16
+}
+
+func (s *topicErrorLeaseStore) Metadata(ctx context.Context, topics []string) (*metadata.ClusterMetadata, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	meta := &metadata.ClusterMetadata{}
+	for _, name := range topics {
+		meta.Topics = append(meta.Topics, protocol.MetadataTopic{Topic: kmsg.StringPtr(name), ErrorCode: s.code})
+	}
+	return meta, nil
+}
+func TestProducePreservesTopicMetadataErrorsAcrossLeaseOwners(t *testing.T) {
+	store, a, b := leaseMetadataErrorBrokers(t)
+	// The first broker holds the lease for a partition absent from metadata.
+	raw := producePromptly(t, a, unknownPartitionProduce("orders", 7, -1, 0))
+	if p := decodeKmsgResponse(t, 3, raw, kmsg.NewPtrProduceResponse).Topics[0].Partitions[0]; p.ErrorCode != protocol.UNKNOWN_TOPIC_OR_PARTITION {
+		t.Fatal(p.ErrorCode)
+	}
+	for _, code := range []int16{protocol.UNKNOWN_TOPIC_OR_PARTITION, protocol.INVALID_TOPIC_EXCEPTION, protocol.TOPIC_AUTHORIZATION_FAILED, protocol.LEADER_NOT_AVAILABLE, protocol.REQUEST_TIMED_OUT, protocol.UNKNOWN_SERVER_ERROR} {
+		t.Run(fmt.Sprintf("code=%d", code), func(t *testing.T) {
+			b.store = &topicErrorLeaseStore{EtcdStore: store, code: code}
+			raw := producePromptly(t, b, unknownPartitionProduce("orders", 7, -1, 0))
+			p := decodeKmsgResponse(t, 3, raw, kmsg.NewPtrProduceResponse).Topics[0].Partitions[0]
+			if p.ErrorCode != code {
+				t.Fatalf("topic metadata error %d became %d", code, p.ErrorCode)
+			}
+		})
+	}
+}
diff --git a/cmd/broker/produce_unknown_partition_test.go b/cmd/broker/produce_unknown_partition_test.go
new file mode 100644
index 00000000..0205a4d5
--- /dev/null
+++ b/cmd/broker/produce_unknown_partition_test.go
@@ -0,0 +1,342 @@
+// Copyright 2025 Alexander Alten (novatechflow), NovaTechflow (novatechflow.com).
+// This project is supported and financed by Scalytics, Inc. (www.scalytics.io).
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/KafScale/platform/pkg/metadata"
+	"github.com/KafScale/platform/pkg/protocol"
+	"github.com/KafScale/platform/pkg/storage"
+	"github.com/twmb/franz-go/pkg/kmsg"
+)
+
+type partitionLookupStore struct {
+	metadata.Store
+	reads      atomic.Int32
+	creates    atomic.Int32
+	createRace bool
+	stale      bool
+}
+
+func (s *partitionLookupStore) NextOffset(ctx context.Context, topic string, partition int32) (int64, error) {
+	s.reads.Add(1)
+	if err := ctx.Err(); err != nil {
+		return 0, err
+	}
+	if s.stale {
+		return 0, metadata.ErrUnknownTopic
+	}
+	return s.Store.NextOffset(ctx, topic, partition)
+}
+func (s *partitionLookupStore) CreateTopic(ctx context.Context, spec metadata.TopicSpec) (*protocol.MetadataTopic, error) {
+	s.creates.Add(1)
+	if s.createRace {
+		s.createRace = false
+		_, err := s.Store.CreateTopic(ctx, metadata.TopicSpec{Name: spec.Name, NumPartitions: 1, ReplicationFactor: 1})
+		if err != nil {
+			return nil, err
+		}
+		return nil, metadata.ErrTopicExists
+	}
+	return s.Store.CreateTopic(ctx, spec)
+}
+
+func producePromptly(t *testing.T, h *handler, req *kmsg.ProduceRequest) []byte {
+	t.Helper()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	type result struct {
+		raw []byte
+		err error
+	}
+	done := make(chan result, 1)
+	go func() {
+		raw, err := h.handleProduce(ctx, &protocol.RequestHeader{APIVersion: 3}, req)
+		done <- result{raw, err}
+	}()
+	select {
+	case out := <-done:
+		if out.err != nil {
+			t.Fatal(out.err)
+		}
+		return out.raw
+	case <-time.After(3 * time.Second):
+		cancel()
+		select {
+		case <-done:
+		case <-time.After(time.Second):
+			t.Fatal("produce did not stop after cancellation")
+		}
+		t.Fatal("produce to a missing partition waited instead of returning promptly")
+		return nil
+	}
+}
+func unknownPartitionProduce(topic string, partition int32, acks int16, timeout int32) *kmsg.ProduceRequest {
+	return &kmsg.ProduceRequest{Acks: acks, TimeoutMillis: timeout, Topics: []kmsg.ProduceRequestTopic{{Topic: topic, Partitions: []kmsg.ProduceRequestTopicPartition{{Partition: partition, Records: testBatchBytes(0, 0, 1)}}}}}
+}
+func TestProduceRejectsUnknownPartitionPromptly(t *testing.T) {
+	for _, partition := range []int32{-1, 1, 7, 1<<31 - 1} {
+		for _, acks := range []int16{-1, 1, 0} {
+			for _, timeout := range []int32{0, 30000} {
+				t.Run(fmt.Sprintf("partition=%d/acks=%d/timeout=%d", partition, acks, timeout), func(t *testing.T) {
+					base := metadata.NewInMemoryStore(defaultMetadata())
+					store := &partitionLookupStore{Store: base}
+					h := newTestHandler(store)
+					t.Cleanup(h.coordinator.Stop)
+					h.autoCreateTopics = true
+					raw := producePromptly(t, h, unknownPartitionProduce("orders", partition, acks, timeout))
+					if acks == 0 {
+						if raw != nil {
+							t.Fatal("acks=0 returned a response")
+						}
+					} else {
+						p := decodeKmsgResponse(t, 3, raw, kmsg.NewPtrProduceResponse).Topics[0].Partitions[0]
+						if p.Partition != partition || p.ErrorCode != protocol.UNKNOWN_TOPIC_OR_PARTITION {
+							t.Fatalf("partition error=%d, want UNKNOWN_TOPIC_OR_PARTITION", p.ErrorCode)
+						}
+					}
+					if store.creates.Load() > 1 || store.reads.Load() > 2 {
+						t.Fatalf("repeated auto-creation: creates=%d reads=%d", store.creates.Load(), store.reads.Load())
+					}
+					meta, err := base.Metadata(context.Background(), []string{"orders"})
+					if err != nil {
+						t.Fatal(err)
+					}
+					if len(meta.Topics[0].Partitions) != 1 {
+						t.Fatal("invalid produce expanded existing topic")
+					}
+					if next, err := base.NextOffset(context.Background(), "orders", 0); err != nil || next != 0 {
+						t.Fatalf("valid partition changed: %d %v", next, err)
+					}
+					if len(h.logs) != 0 {
+						t.Fatal("missing partition initialized a log")
+					}
+				})
+			}
+		}
+	}
+}
+
+func TestProduceUnknownPartitionWithoutAutoCreate(t *testing.T) {
+	store := &partitionLookupStore{Store: metadata.NewInMemoryStore(defaultMetadata())}
+	h := newTestHandler(store)
+	t.Cleanup(h.coordinator.Stop)
+	h.autoCreateTopics = false
+	raw := producePromptly(t, h, unknownPartitionProduce("orders", 7, -1, 0))
+	if p := decodeKmsgResponse(t, 3, raw, kmsg.NewPtrProduceResponse).Topics[0].Partitions[0]; p.ErrorCode != protocol.UNKNOWN_TOPIC_OR_PARTITION {
+		t.Fatal(p.ErrorCode)
+	}
+	if store.creates.Load() != 0 {
+		t.Fatal("auto-created with feature disabled")
+	}
+}
+func TestProduceContinuesPastUnknownPartition(t *testing.T) {
+	store := metadata.NewInMemoryStore(defaultMetadata())
+	h := newTestHandler(store)
+	t.Cleanup(h.coordinator.Stop)
+	h.autoCreateTopics = true
+	req := unknownPartitionProduce("orders", 7, -1, 0)
+	req.Topics[0].Partitions = append(req.Topics[0].Partitions, kmsg.ProduceRequestTopicPartition{Partition: 0, Records: testBatchBytes(0, 0, 1)})
+	raw := producePromptly(t, h, req)
+	parts := decodeKmsgResponse(t, 3, raw, kmsg.NewPtrProduceResponse).Topics[0].Partitions
+	if len(parts) != 2 || parts[0].ErrorCode != protocol.UNKNOWN_TOPIC_OR_PARTITION || parts[1].ErrorCode != protocol.NONE || parts[1].BaseOffset != 0 {
+		t.Fatalf("unexpected partition results: %+v", parts)
+	}
+	if next, err := store.NextOffset(context.Background(), "orders", 0); err != nil || next != 1 {
+		t.Fatalf("valid partition did not advance: %d %v", next, err)
+	}
+}
+func TestProduceAutoCreateRaceRejectsMissingPartition(t *testing.T) {
+	base := metadata.NewInMemoryStore(defaultMetadata())
+	store := &partitionLookupStore{Store: base, createRace: true}
+	h := newTestHandler(store)
+	t.Cleanup(h.coordinator.Stop)
+	h.autoCreateTopics = true
+	raw := producePromptly(t, h, unknownPartitionProduce("raced", 7, -1, 0))
+	if p := decodeKmsgResponse(t, 3, raw, kmsg.NewPtrProduceResponse).Topics[0].Partitions[0]; p.ErrorCode != protocol.UNKNOWN_TOPIC_OR_PARTITION {
+		t.Fatal(p.ErrorCode)
+	}
+	if store.creates.Load() != 1 || store.reads.Load() != 2 {
+		t.Fatalf("unbounded retry after concurrent create: creates=%d reads=%d", store.creates.Load(), store.reads.Load())
+	}
+	meta, err := base.Metadata(context.Background(), []string{"raced"})
+	if err != nil || len(meta.Topics) != 1 || len(meta.Topics[0].Partitions) != 1 {
+		t.Fatal("changed concurrent creator's topic", err)
+	}
+}
+func TestProduceAutoCreateRetriesOnceWhenMetadataStaysMissing(t *testing.T) {
+	store := &partitionLookupStore{Store: metadata.NewInMemoryStore(defaultMetadata()), stale: true}
+	h := newTestHandler(store)
+	t.Cleanup(h.coordinator.Stop)
+	h.autoCreateTopics = true
+	raw := producePromptly(t, h, unknownPartitionProduce("new-topic", 0, -1, 0))
+	if p := decodeKmsgResponse(t, 3, raw, kmsg.NewPtrProduceResponse).Topics[0].Partitions[0]; p.ErrorCode != protocol.UNKNOWN_TOPIC_OR_PARTITION {
+		t.Fatal(p.ErrorCode)
+	}
+	if store.creates.Load() != 1 || store.reads.Load() != 2 {
+		t.Fatalf("metadata retry did not stop: creates=%d reads=%d", store.creates.Load(), store.reads.Load())
+	}
+}
+func TestProduceAutoCreatesMissingTopicWithRequestedPartition(t *testing.T) {
+	store := metadata.NewInMemoryStore(defaultMetadata())
+	h := newTestHandler(store)
+	t.Cleanup(h.coordinator.Stop)
+	h.autoCreateTopics = true
+	raw := producePromptly(t, h, unknownPartitionProduce("new-topic", 2, -1, 0))
+	if p := decodeKmsgResponse(t, 3, raw, kmsg.NewPtrProduceResponse).Topics[0].Partitions[0]; p.ErrorCode != protocol.NONE || p.BaseOffset != 0 {
+		t.Fatalf("auto-create failed: %+v", p)
+	}
+	if next, err := store.NextOffset(context.Background(), "new-topic", 2); err != nil || next != 1 {
+		t.Fatalf("created partition not written: %d %v", next, err)
+	}
+}
+func TestProduceRejectsUnknownPartitionWithEtcdLeases(t *testing.T) {
+	store := newEtcdTopicStore(t, context.Background())
+	h := newTestHandler(store)
+	t.Cleanup(h.coordinator.Stop)
+	t.Cleanup(h.leaseManager.ReleaseAll)
+	h.autoCreateTopics = true
+	raw := producePromptly(t, h, unknownPartitionProduce("orders", 7, -1, 0))
+	if p := decodeKmsgResponse(t, 3, raw, kmsg.NewPtrProduceResponse).Topics[0].Partitions[0]; p.ErrorCode != protocol.UNKNOWN_TOPIC_OR_PARTITION {
+		t.Fatal(p.ErrorCode)
+	}
+	if len(h.logs) != 0 {
+		t.Fatal("missing partition initialized a log with leases enabled")
+	}
+}
+
+type staleLeaseMetadataStore struct {
+	*metadata.EtcdStore
+	refreshed  atomic.Bool
+	refreshes  atomic.Int32
+	refreshErr error
+}
+
+func (s *staleLeaseMetadataStore) Metadata(ctx context.Context, topics []string) (*metadata.ClusterMetadata, error) {
+	if !s.refreshed.Load() {
+		return &metadata.ClusterMetadata{}, nil
+	}
+	return s.EtcdStore.Metadata(ctx, topics)
+}
+func (s *staleLeaseMetadataStore) RefreshSnapshot(ctx context.Context) error {
+	s.refreshes.Add(1)
+	if s.refreshErr != nil {
+		return s.refreshErr
+	}
+	if err := s.EtcdStore.RefreshSnapshot(ctx); err != nil {
+		return err
+	}
+	s.refreshed.Store(true)
+	return nil
+}
+
+func TestProduceUnknownPartitionWithAnotherLeaseOwner(t *testing.T) {
+	store := newEtcdTopicStore(t, context.Background())
+	a := newTestHandler(store)
+	b := newHandler(store, storage.NewMemoryS3Client(), protocol.MetadataBroker{NodeID: 2}, testLogger())
+	for _, h := range []*handler{a, b} {
+		t.Cleanup(h.coordinator.Stop)
+		t.Cleanup(h.leaseManager.ReleaseAll)
+		h.autoCreateTopics = true
+	}
+	for _, h := range []*handler{a, b} {
+		raw := producePromptly(t, h, unknownPartitionProduce("orders", 7, -1, 0))
+		p := decodeKmsgResponse(t, 3, raw, kmsg.NewPtrProduceResponse).Topics[0].Partitions[0]
+		if p.ErrorCode != protocol.UNKNOWN_TOPIC_OR_PARTITION {
+			t.Fatalf("broker %d returned %d for a nonexistent partition", h.brokerInfo.NodeID, p.ErrorCode)
+		}
+	}
+
+	// A real partition on the other broker must still produce a leader retry.
+	raw := producePromptly(t, a, unknownPartitionProduce("orders", 0, -1, 0))
+	if p := decodeKmsgResponse(t, 3, raw, kmsg.NewPtrProduceResponse).Topics[0].Partitions[0]; p.ErrorCode != protocol.NONE {
+		t.Fatal(p.ErrorCode)
+	}
+	stale := &staleLeaseMetadataStore{EtcdStore: store}
+	b.store = stale
+	req := unknownPartitionProduce("orders", 0, -1, 0)
+	req.Topics[0].Partitions = append(req.Topics[0].Partitions, unknownPartitionProduce("orders", 7, -1, 0).Topics[0].Partitions[0])
+	raw = producePromptly(t, b, req)
+	parts := decodeKmsgResponse(t, 3, raw, kmsg.NewPtrProduceResponse).Topics[0].Partitions
+	if len(parts) != 2 || parts[0].ErrorCode != protocol.NOT_LEADER_OR_FOLLOWER || parts[1].ErrorCode != protocol.UNKNOWN_TOPIC_OR_PARTITION {
+		t.Fatalf("mixed lease errors were misclassified: %+v", parts)
+	}
+	if !stale.refreshed.Load() || stale.refreshes.Load() != 1 {
+		t.Fatal("lease errors did not use one fresh metadata snapshot")
+	}
+	if next, err := store.NextOffset(context.Background(), "orders", 0); err != nil || next != 1 {
+		t.Fatalf("non-owner changed valid offsets: %d %v", next, err)
+	}
+	// A failed refresh cannot establish that a partition is absent.
+	failed := &staleLeaseMetadataStore{EtcdStore: store, refreshErr: fmt.Errorf("snapshot unavailable")}
+	b.store = failed
+	raw = producePromptly(t, b, unknownPartitionProduce("orders", 7, -1, 0))
+	if p := decodeKmsgResponse(t, 3, raw, kmsg.NewPtrProduceResponse).Topics[0].Partitions[0]; p.ErrorCode != protocol.NOT_LEADER_OR_FOLLOWER {
+		t.Fatalf("failed metadata refresh changed the known ownership error: %d", p.ErrorCode)
+	}
+	if failed.refreshes.Load() != 1 {
+		t.Fatal("failed metadata refresh was not exercised")
+	}
+}
+
+func TestProduceConcurrentAutoCreateWithRequestedPartition(t *testing.T) {
+	const producers = 32
+	store := metadata.NewInMemoryStore(defaultMetadata())
+	h := newTestHandler(store)
+	t.Cleanup(h.coordinator.Stop)
+	h.autoCreateTopics = true
+	h.flushOnAck = true
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	type result struct {
+		raw []byte
+		err error
+	}
+	done := make(chan result, producers)
+	for range producers {
+		go func() {
+			raw, err := h.handleProduce(ctx, &protocol.RequestHeader{APIVersion: 3}, unknownPartitionProduce("concurrent-topic", 2, -1, 0))
+			done <- result{raw, err}
+		}()
+	}
+	offsets := make(map[int64]bool)
+	for range producers {
+		select {
+		case out := <-done:
+			if out.err != nil {
+				t.Fatal(out.err)
+			}
+			p := decodeKmsgResponse(t, 3, out.raw, kmsg.NewPtrProduceResponse).Topics[0].Partitions[0]
+			if p.ErrorCode != protocol.NONE {
+				t.Fatalf("concurrent auto-create rejected valid partition: %d", p.ErrorCode)
+			}
+			if offsets[p.BaseOffset] {
+				t.Fatalf("duplicate acknowledged offset %d", p.BaseOffset)
+			}
+			offsets[p.BaseOffset] = true
+		case <-ctx.Done():
+			t.Fatal("concurrent auto-creation did not finish", ctx.Err())
+		}
+	}
+	if next, err := store.NextOffset(context.Background(), "concurrent-topic", 2); err != nil || next != producers {
+		t.Fatalf("lost acknowledged offsets: %d %v", next, err)
+	}
+}
diff --git a/docs/releases/unreleased.md b/docs/releases/unreleased.md
index 8327e2ff..d35a8a8d 100644
--- a/docs/releases/unreleased.md
+++ b/docs/releases/unreleased.md
@@ -37,6 +37,8 @@ limitations under the License.
 
 ## Bug fixes
 
+- Produce requests targeting a partition that does not exist now finish promptly. Auto-creation makes at most one attempt per partition-log lookup instead of spinning when the topic already exists. Requests requiring a response receive `UNKNOWN_TOPIC_OR_PARTITION`, including when another broker holds a lease for that nonexistent partition; valid partitions still receive normal leader redirection. Topic-level metadata errors, including invalid topic names, are preserved across lease owners. Requests with `acks=0` still send no response. Requests with no timeout no longer occupy a broker worker indefinitely, and other valid partitions in the same produce request continue to be processed (#326).
+
 - Broker topic creation reads the authoritative etcd metadata snapshot and uses compare-and-swap to preserve concurrent changes. Racing creates no longer lose a topic, and a stale broker cannot replace the cluster's brokers or existing partition counts with its startup view. Startup demo topics are never copied into etcd. Partition expansion and topic deletion use the same snapshot fencing; expansion publishes missing partition state atomically in batches without resetting existing state, and a stale operation cannot modify a deleted and recreated topic. Snapshot watches retry failed initial reads and replay updates committed between the full read and watch registration. Operator publication rebuilds each CAS retry from its original desired state and the latest snapshot, preventing deleted auto-created topics from reappearing; unreadable snapshots are preserved and reported as errors (#325).
 
 - `make lfs-demo` no longer loses LFS tracker events. The demo brokers advertised the host port forward address `127.0.0.1:39092`, which the LFS proxy pod cannot reach, so the tracker could not create `__lfs_ops_state` or deliver events and the proxy logged a warning. The brokers now advertise `kafscale-broker.kafscale-demo.svc.cluster.local:9092`. The LFS proxy advertises `127.0.0.1:39092`, and `lfs-client-sdk` `pf-start` forwards that port to the proxy instead of the broker, so host clients connect through the proxy. The proxy no longer logs warnings in this stack (#308).