allow set force partition for topic writer

(cherry picked from commit a699028)

Author: rekby

internal/topic/topicwriterinternal/writer_reconnector.go

@@ -66,7 +66,7 @@ type WriterReconnectorConfig struct {
 }
 
 func (cfg *WriterReconnectorConfig) validate() error {
-	if cfg.producerID != cfg.defaultPartitioning.MessageGroupID {
+	if cfg.defaultPartitioning.Type == rawtopicwriter.PartitioningMessageGroupID && cfg.producerID != cfg.defaultPartitioning.MessageGroupID {
 		return xerrors.WithStackTrace(errProducerIDNotEqualMessageGroupID)
 	}
 	return nil

tests/integration/topic_read_writer_test.go

@@ -388,6 +388,22 @@ func TestUpdateToken(t *testing.T) {
 	xtest.WaitChannelClosed(t, activityStopped)
 }
 
+func TestTopicWriterWithManualPartitionSelect(t *testing.T) {
+	ctx := xtest.Context(t)
+	db := connect(t)
+	topicPath := createTopic(ctx, t, db)
+
+	writer, err := db.Topic().StartWriter(
+		"producer-id",
+		topicPath,
+		topicoptions.WithPartitionID(0),
+		topicoptions.WithSyncWrite(true),
+	)
+	require.NoError(t, err)
+	err = writer.Write(ctx, topicwriter.Message{Data: strings.NewReader("asd")})
+	require.NoError(t, err)
+}
+
 var topicCounter int
 
 func createTopic(ctx context.Context, t testing.TB, db ydb.Connection) (topicPath string) {