[CELEBORN-2481] Fix NPE when persisting MapFileMeta without segment granularity

[attiasr]

What changes were proposed in this pull request?

PbSerDeUtils.toPbFileInfo now adds the MAP partition's writing-segment map and segment indexes only when they are non-null.

MapFileMeta creates both lazily, on the first addPartitionSegmentId / addSegmentIdAndFirstBufferIndex call. The worker builds MAP file metas with new MapFileMeta(), so for a MAP partition written without segment granularity both fields stay null, and putAllPartitionWritingSegment(null) threw:

java.lang.NullPointerException: Cannot invoke "java.util.Map.keySet()" because "m" is null
  at com.google.protobuf.MapField$MutabilityAwareMap.putAll(MapField.java:312)
  at org.apache.celeborn.common.protocol.PbFileInfo$Builder.putAllPartitionWritingSegment(PbFileInfo.java:1674)
  at org.apache.celeborn.common.util.PbSerDeUtils$.toPbFileInfo(PbSerDeUtils.scala:164)

On restore, fromPbFileInfo restores both fields as null when they are empty, matching a live meta that never recorded a segment, so MapFileMeta#hasPartitionSegmentIds() gives the same answer before and after a restart. Restored metas are already marked writer-closed, so nothing writes to them. Partitions with segment granularity serialize and restore exactly as before.

Why are the changes needed?

When graceful shutdown is enabled, the worker persists committed file infos through toPbFileInfoMap → toPbFileInfo in two places, and the NPE breaks both:

• The periodic saver (worker-storage-manager-committed-fileinfo-saver). It runs with scheduleWithFixedDelay and does not catch the NPE, so the first failure cancels it and nothing is saved afterwards.
• saveAllCommittedFileInfosToDB at shutdown. It logs the error and skips any shuffle that contains such a MAP partition.

As a result, after a graceful restart the worker cannot recover MAP partitions written without segment granularity, such as Flink shuffle partitions. Readers fail to find them, and the producing tasks have to be re-run.

Does this PR resolve a correctness bug?

• Yes

Shuffle data is never incorrect. Partitions are lost on graceful restart instead of being recovered.

Does this PR introduce any user-facing change?

• Yes

With graceful shutdown enabled, a worker restart now recovers committed MAP partitions without segment granularity. No configuration or protocol change.

How was this patch tested?

Added two tests to PbSerDeUtilsTest:

• fromAndToPbFileInfo with MapFileMeta without segment granularity builds the meta the way the worker does (no-arg constructor plus buffer size and subpartition count). It round-trips it through toPbFileInfo/fromPbFileInfo and through toPbFileInfoMap/fromPbFileInfoMap, checking the buffer size, subpartition count, segment-granularity flag, writer-closed flag, that hasPartitionSegmentIds() is false and the segment indexes are null, a segment lookup, path and length. On main it fails with the NPE above.
• fromAndToPbFileInfo with MapFileMeta with segment granularity is a regression check. It confirms that hasPartitionSegmentIds() stays true and that recorded segment IDs and segment indexes still round-trip.

build/mvn -pl common -am -Dtest=none -DwildcardSuites=org.apache.celeborn.common.util.PbSerDeUtilsTest test
Tests: succeeded 36, failed 0

[yew1eb]

Verified the NPE and that the periodic saver (saveCommittedFileInfosExecutor) silently stops after the first uncaught exception. LGTM.

One note: after restore these fields are empty collections rather than null, so MapFileMeta#hasPartitionSegmentIds() now returns true for restored non-segment MAP partitions. That is harmless today since reader selection is based on isSegmentGranularityVisible(), but it is a subtle semantic difference worth a sentence in the PR description.

[attiasr]

Good point, thanks. Rather than documenting the difference, I made fromPbFileInfo restore empty segment fields as null (4c2784e), so hasPartitionSegmentIds() on a restored non-segment MAP meta matches a live one. The round-trip tests now assert it in both directions: false without segment granularity, true with it.

[yew1eb]

+1, LGTM. Verified against main: both fields are lazily initialized in MapFileMeta, and the worker builds MAP file metas via the no-arg constructor in both createMemoryFileInfo and getFileMeta, so both stay null without segment granularity. The restore path is also safe — restored metas are writer-closed and non-segment MAP partitions are read through plain MapPartitionData.

One process nit: the PR title is missing a JIRA id — Apache projects generally expect [CELEBORN-xxxx] in the title. Could you file a JIRA and retitle?

[attiasr]

Thanks for the review @yew1eb.
I've opened a ticket and renamed the PR