[FLINK-17593] Turn BucketStateSerializerTest into an upgrade test

We need this for future changes to the serialization format.

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/Bucket.java

@@ -269,7 +269,14 @@ public class Bucket<IN, BucketID> {
 			}
 		}
 
-		return new BucketState<>(bucketId, bucketPath, inProgressFileCreationTime, inProgressResumable, pendingPartsPerCheckpoint);
+		// It's important to create a defensive copy here! Otherwise, later calls to
+		// onSuccessfulCompletionOfCheckpoint would change a BucketState that we already gave out.
+		final NavigableMap<Long, List<CommitRecoverable>> copiedPendingPartsPerCheckpoint = new TreeMap<>();
+		for (Map.Entry<Long, List<CommitRecoverable>> checkpoint : pendingPartsPerCheckpoint.entrySet()) {
+			copiedPendingPartsPerCheckpoint.put(checkpoint.getKey(), new ArrayList<>(checkpoint.getValue()));
+		}
+
+		return new BucketState<>(bucketId, bucketPath, inProgressFileCreationTime, inProgressResumable, copiedPendingPartsPerCheckpoint);
 	}
 
 	private void prepareBucketForCheckpointing(long checkpointId) throws IOException {

flink-streaming-java/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1/bucket/test-bucket/part-0-0

+wrote
+wrote

flink-streaming-java/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1/bucket/test-bucket/part-0-1

+wrote
+wrote

flink-streaming-java/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1/bucket/test-bucket/part-0-2

+wrote
+wrote

flink-streaming-java/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1/bucket/test-bucket/part-0-3

+wrote
+wrote

flink-streaming-java/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1/bucket/test-bucket/part-0-4

+wrote
+wrote

flink-streaming-java/src/test/resources/bucket-state-migration-test/full-v1/bucket/test-bucket/.part-0-5.inprogress.795d25e1-c9b1-4635-afc1-9334651de3c9

+writing

flink-streaming-java/src/test/resources/bucket-state-migration-test/full-v1/bucket/test-bucket/part-0-0

+wrote
+wrote

flink-streaming-java/src/test/resources/bucket-state-migration-test/full-v1/bucket/test-bucket/part-0-1

+wrote
+wrote

flink-streaming-java/src/test/resources/bucket-state-migration-test/full-v1/bucket/test-bucket/part-0-2

+wrote
+wrote

flink-streaming-java/src/test/resources/bucket-state-migration-test/full-v1/bucket/test-bucket/part-0-3

+wrote
+wrote

flink-streaming-java/src/test/resources/bucket-state-migration-test/full-v1/bucket/test-bucket/part-0-4

+wrote
+wrote

flink-streaming-java/src/test/resources/bucket-state-migration-test/only-in-progress-v1/bucket/test-bucket/.part-0-0.inprogress.0ba167b7-76c7-4ba8-9e83-80c6000b5320

+writing

pom.xml

@@ -1458,6 +1458,7 @@ under the License.
 						<exclude>**/src/test/resources/*-snapshot</exclude>
 						<exclude>**/src/test/resources/*.snapshot</exclude>
 						<exclude>**/src/test/resources/*-savepoint</exclude>
+						<exclude>**/src/test/resources/bucket-state-migration-test/**</exclude>
 						<exclude>flink-core/src/test/resources/serialized-kryo-serializer-1.3</exclude>
 						<exclude>flink-core/src/test/resources/type-without-avro-serialized-using-kryo</exclude>
 						<exclude>flink-formats/flink-avro/src/test/resources/flink-1.4-serializer-java-serialized</exclude>