changefeedccl: kafka changefeeds may send resolved messages to only some partitions about cockroach HOT 2 OPEN

diff --git a/pkg/ccl/changefeedccl/sink_kafka.go b/pkg/ccl/changefeedccl/sink_kafka.go
index 69194e7018b..d64e30065cf 100644
--- a/pkg/ccl/changefeedccl/sink_kafka.go
+++ b/pkg/ccl/changefeedccl/sink_kafka.go
@@ -377,6 +377,10 @@ func (s *kafkaSink) EmitRow(
        return s.emitMessage(ctx, msg)
 }
 
+type resolvedMetadata struct {
+       partition int32
+}
+
 // EmitResolvedTimestamp implements the Sink interface.
 func (s *kafkaSink) EmitResolvedTimestamp(
        ctx context.Context, encoder Encoder, resolved hlc.Timestamp,
@@ -419,6 +423,7 @@ func (s *kafkaSink) EmitResolvedTimestamp(
                                Partition: partition,
                                Key:       nil,
                                Value:     sarama.ByteEncoder(payload),
+                               Metadata:  resolvedMetadata{partition: partition},
                        }
                        if err := s.emitMessage(ctx, msg); err != nil {
                                return err
@@ -622,6 +627,11 @@ func (s *kafkaSink) finishProducerMessage(ackMsg *sarama.ProducerMessage, ackErr
                }
                m.alloc.Release(s.ctx)
        }
+       if m, ok := ackMsg.Metadata.(resolvedMetadata); ok {
+               if m.partition != ackMsg.Partition {
+                       panic(fmt.Sprintf("partition mismatch, requested %d, sent to %d", m.partition, ackMsg.Partition))
+               }
+       }
        if s.mu.flushErr == nil && ackError != nil {
                s.mu.flushErr = ackError
        }
diff --git a/pkg/cmd/roachtest/tests/cdc.go b/pkg/cmd/roachtest/tests/cdc.go
index b419c6bb21d..710bd9e5d2a 100644
--- a/pkg/cmd/roachtest/tests/cdc.go
+++ b/pkg/cmd/roachtest/tests/cdc.go
@@ -1136,7 +1136,10 @@ func registerCDC(r registry.Registry) {
                        feed := ct.newChangefeed(feedArgs{
                                sinkType: kafkaSink,
                                targets:  allTpccTargets,
-                               opts:     map[string]string{"initial_scan": "'no'"},
+                               opts: map[string]string{
+                                       "initial_scan": "'no'",
+                                       "resolved":     "",
+                               },
                        })
                        ct.runFeedLatencyVerifier(feed, latencyTargets{
                                initialScanLatency: 3 * time.Minute,
@@ -2746,7 +2749,7 @@ func (k kafkaManager) createTopic(ctx context.Context, topic string) error {
                        return errors.Wrap(err, "admin client")
                }
                return admin.CreateTopic(topic, &sarama.TopicDetail{
-                       NumPartitions:     1,
+                       NumPartitions:     5,
                        ReplicationFactor: 1,
                }, false)
        })