From 499a83a172e04e117dfb32c1a75905e90ef3a583 Mon Sep 17 00:00:00 2001 From: jiangpengcheng Date: Wed, 8 Jul 2026 17:04:32 +0800 Subject: [PATCH 1/3] Fix schema sidecar atomic write --- pf/schema_sidecar.go | 28 ++++++++++++++++++++++++- pf/schema_sidecar_test.go | 44 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 1 deletion(-) diff --git a/pf/schema_sidecar.go b/pf/schema_sidecar.go index e3aa556..9d945af 100644 --- a/pf/schema_sidecar.go +++ b/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: diff --git a/pf/schema_sidecar_test.go b/pf/schema_sidecar_test.go index 44cd3e9..d977b82 100644 --- a/pf/schema_sidecar_test.go +++ b/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) From 3ed33e161bfae739c31f1637a5e37da63bcdfe13 Mon Sep 17 00:00:00 2001 From: jiangpengcheng Date: Mon, 13 Jul 2026 20:18:19 +0800 Subject: [PATCH 2/3] Preserve schema sidecar permissions --- pf/schema_sidecar.go | 39 +++++++++++++++++++++++++++------------ pf/schema_sidecar_test.go | 23 +++++++++++++++++++++++ 2 files changed, 50 insertions(+), 12 deletions(-) diff --git a/pf/schema_sidecar.go b/pf/schema_sidecar.go index 9d945af..702543f 100644 --- a/pf/schema_sidecar.go +++ b/pf/schema_sidecar.go @@ -115,26 +115,41 @@ func writeSinkSchemaSidecarToDir(workDir string, schema SinkSchema) (string, err } func writeFileAtomic(path string, content []byte, perm os.FileMode) error { - tmpFile, err := os.CreateTemp(filepath.Dir(path), filepath.Base(path)+".*.tmp") + var existingPerm os.FileMode + if info, err := os.Lstat(path); err == nil { + if info.Mode().IsRegular() { + existingPerm = info.Mode().Perm() + } + } else if !os.IsNotExist(err) { + return err + } + + tmpDir, err := os.MkdirTemp(filepath.Dir(path), filepath.Base(path)+".*.tmp") if err != nil { return err } - tmpPath := tmpFile.Name() defer func() { - _ = os.Remove(tmpPath) + _ = os.RemoveAll(tmpDir) }() - _, writeErr := tmpFile.Write(content) - chmodErr := tmpFile.Chmod(perm) - closeErr := tmpFile.Close() - if writeErr != nil { - return writeErr + tmpPath := filepath.Join(tmpDir, filepath.Base(path)) + tmpFile, err := os.OpenFile(tmpPath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, perm) + if err != nil { + return err } - if chmodErr != nil { - return chmodErr + + if _, err := tmpFile.Write(content); err != nil { + _ = tmpFile.Close() + return err } - if closeErr != nil { - return closeErr + if existingPerm != 0 { + if err := tmpFile.Chmod(existingPerm); err != nil { + _ = tmpFile.Close() + return err + } + } + if err := tmpFile.Close(); err != nil { + return err } return os.Rename(tmpPath, path) diff --git a/pf/schema_sidecar_test.go b/pf/schema_sidecar_test.go index d977b82..f7ac6cd 100644 --- a/pf/schema_sidecar_test.go +++ b/pf/schema_sidecar_test.go @@ -177,6 +177,29 @@ func TestWriteSinkSchemaSidecarToDirAtomicallyReplacesExistingSidecar(t *testing } } +func TestWriteSinkSchemaSidecarToDirPreservesExistingSidecarPermissions(t *testing.T) { + workDir := t.TempDir() + sidecarFile := filepath.Join(workDir, SinkSchemaSidecarFileName) + if err := os.WriteFile(sidecarFile, []byte("old"), 0600); err != nil { + t.Fatalf("write existing sidecar: %v", err) + } + if err := os.Chmod(sidecarFile, 0600); err != nil { + t.Fatalf("chmod existing sidecar: %v", err) + } + + if _, err := writeSinkSchemaSidecarToDir(workDir, SinkSchema{SchemaType: SchemaTypeJSON}); err != nil { + t.Fatalf("writeSinkSchemaSidecarToDir returned error: %v", err) + } + + info, err := os.Stat(sidecarFile) + if err != nil { + t.Fatalf("stat sidecar: %v", err) + } + if got, want := info.Mode().Perm(), os.FileMode(0600); got != want { + t.Fatalf("sidecar permissions = %04o, want %04o", got, want) + } +} + func TestSchemaTypeBoolUsesPulsarBooleanName(t *testing.T) { if SchemaTypeBool != "boolean" { t.Fatalf("SchemaTypeBool = %q, want boolean", SchemaTypeBool) From fbb3622bb3f6813bcf7be3d9b6c5c97768e62b27 Mon Sep 17 00:00:00 2001 From: jiangpengcheng Date: Mon, 13 Jul 2026 20:19:07 +0800 Subject: [PATCH 3/3] Handle zero schema sidecar mode --- pf/schema_sidecar.go | 4 +++- pf/schema_sidecar_test.go | 38 +++++++++++++++++++++----------------- 2 files changed, 24 insertions(+), 18 deletions(-) diff --git a/pf/schema_sidecar.go b/pf/schema_sidecar.go index 702543f..42f83ed 100644 --- a/pf/schema_sidecar.go +++ b/pf/schema_sidecar.go @@ -116,9 +116,11 @@ func writeSinkSchemaSidecarToDir(workDir string, schema SinkSchema) (string, err func writeFileAtomic(path string, content []byte, perm os.FileMode) error { var existingPerm os.FileMode + var hasExistingPerm bool if info, err := os.Lstat(path); err == nil { if info.Mode().IsRegular() { existingPerm = info.Mode().Perm() + hasExistingPerm = true } } else if !os.IsNotExist(err) { return err @@ -142,7 +144,7 @@ func writeFileAtomic(path string, content []byte, perm os.FileMode) error { _ = tmpFile.Close() return err } - if existingPerm != 0 { + if hasExistingPerm { if err := tmpFile.Chmod(existingPerm); err != nil { _ = tmpFile.Close() return err diff --git a/pf/schema_sidecar_test.go b/pf/schema_sidecar_test.go index f7ac6cd..e4d9688 100644 --- a/pf/schema_sidecar_test.go +++ b/pf/schema_sidecar_test.go @@ -178,25 +178,29 @@ func TestWriteSinkSchemaSidecarToDirAtomicallyReplacesExistingSidecar(t *testing } func TestWriteSinkSchemaSidecarToDirPreservesExistingSidecarPermissions(t *testing.T) { - workDir := t.TempDir() - sidecarFile := filepath.Join(workDir, SinkSchemaSidecarFileName) - if err := os.WriteFile(sidecarFile, []byte("old"), 0600); err != nil { - t.Fatalf("write existing sidecar: %v", err) - } - if err := os.Chmod(sidecarFile, 0600); err != nil { - t.Fatalf("chmod existing sidecar: %v", err) - } + for _, perm := range []os.FileMode{0000, 0600} { + t.Run(perm.String(), func(t *testing.T) { + workDir := t.TempDir() + sidecarFile := filepath.Join(workDir, SinkSchemaSidecarFileName) + if err := os.WriteFile(sidecarFile, []byte("old"), 0600); err != nil { + t.Fatalf("write existing sidecar: %v", err) + } + if err := os.Chmod(sidecarFile, perm); err != nil { + t.Fatalf("chmod existing sidecar: %v", err) + } - if _, err := writeSinkSchemaSidecarToDir(workDir, SinkSchema{SchemaType: SchemaTypeJSON}); err != nil { - t.Fatalf("writeSinkSchemaSidecarToDir returned error: %v", err) - } + if _, err := writeSinkSchemaSidecarToDir(workDir, SinkSchema{SchemaType: SchemaTypeJSON}); err != nil { + t.Fatalf("writeSinkSchemaSidecarToDir returned error: %v", err) + } - info, err := os.Stat(sidecarFile) - if err != nil { - t.Fatalf("stat sidecar: %v", err) - } - if got, want := info.Mode().Perm(), os.FileMode(0600); got != want { - t.Fatalf("sidecar permissions = %04o, want %04o", got, want) + info, err := os.Stat(sidecarFile) + if err != nil { + t.Fatalf("stat sidecar: %v", err) + } + if got := info.Mode().Perm(); got != perm { + t.Fatalf("sidecar permissions = %04o, want %04o", got, perm) + } + }) } }