Fix schema sidecar atomic write

pf/schema_sidecar.go

@@ -108,12 +108,38 @@ func writeSinkSchemaSidecarToDir(workDir string, schema SinkSchema) (string, err
 		return "", err
 	}
 
-	if err := os.WriteFile(sidecarPath, content, 0644); err != nil {
+	if err := writeFileAtomic(sidecarPath, content, 0644); err != nil {
 		return "", err
 	}
 	return sidecarPath, nil
 }
 
+func writeFileAtomic(path string, content []byte, perm os.FileMode) error {
+	tmpFile, err := os.CreateTemp(filepath.Dir(path), filepath.Base(path)+".*.tmp")
+	if err != nil {
+		return err
+	}
+	tmpPath := tmpFile.Name()
+	defer func() {
+		_ = os.Remove(tmpPath)
+	}()
+
+	_, writeErr := tmpFile.Write(content)
+	chmodErr := tmpFile.Chmod(perm)
+	closeErr := tmpFile.Close()
+	if writeErr != nil {
+		return writeErr
+	}
+	if chmodErr != nil {
+		return chmodErr
+	}
+	if closeErr != nil {
+		return closeErr
+	}
+
+	return os.Rename(tmpPath, path)
+}
+
 func isNoSchemaType(schemaType string) bool {
 	switch strings.ToLower(strings.TrimSpace(schemaType)) {
 	case "", SchemaTypeBytes, SchemaTypeNone:

pf/schema_sidecar_test.go

@@ -23,6 +23,7 @@ import (
 	"encoding/json"
 	"os"
 	"path/filepath"
+	"runtime"
 	"testing"
 )
 
@@ -133,6 +134,49 @@ func TestWriteSinkSchemaSidecarToDirTrimsSchemaTypeAndPreservesCasing(t *testing
 	}
 }
 
+func TestWriteSinkSchemaSidecarToDirAtomicallyReplacesExistingSidecar(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("symlink permissions differ on Windows")
+	}
+	workDir := t.TempDir()
+	sidecarFile := filepath.Join(workDir, SinkSchemaSidecarFileName)
+	readOnlyTarget := filepath.Join(workDir, "read-only-target")
+	if err := os.WriteFile(readOnlyTarget, []byte("target"), 0400); err != nil {
+		t.Fatalf("write read-only target: %v", err)
+	}
+	defer func() {
+		_ = os.Chmod(readOnlyTarget, 0600)
+	}()
+	if err := os.Symlink(readOnlyTarget, sidecarFile); err != nil {
+		t.Fatalf("create sidecar symlink: %v", err)
+	}
+
+	sidecarPath, err := writeSinkSchemaSidecarToDir(workDir, SinkSchema{
+		SchemaType: SchemaTypeJSON,
+		SchemaData: `{"type":"record","name":"Student","fields":[]}`,
+	})
+	if err != nil {
+		t.Fatalf("writeSinkSchemaSidecarToDir returned error: %v", err)
+	}
+	if sidecarPath != sidecarFile {
+		t.Fatalf("sidecarPath = %q, want %q", sidecarPath, sidecarFile)
+	}
+	info, err := os.Lstat(sidecarFile)
+	if err != nil {
+		t.Fatalf("lstat sidecar: %v", err)
+	}
+	if info.Mode()&os.ModeSymlink != 0 {
+		t.Fatal("sidecar is still a symlink, want atomically replaced regular file")
+	}
+	targetContent, err := os.ReadFile(readOnlyTarget)
+	if err != nil {
+		t.Fatalf("read target: %v", err)
+	}
+	if string(targetContent) != "target" {
+		t.Fatalf("target content = %q, want unchanged", targetContent)
+	}
+}
+
 func TestSchemaTypeBoolUsesPulsarBooleanName(t *testing.T) {
 	if SchemaTypeBool != "boolean" {
 		t.Fatalf("SchemaTypeBool = %q, want boolean", SchemaTypeBool)