[FLINK-17593][Connectors/FileSystem] Support arbitrary recovery mecha…

…nism for PartFileWriter

This change includes two things:

1. Make the PartFileWriter generic and decouple the PartFileWriter and
RecoverableStream. According to different pre-commit / commit methods,
this change allows us to extend different types of PartFileWriter.

2. Make the Bucket/Buckets depends on the PartFileFactory instead of
RecoverableWriter.

flink-core/src/main/java/org/apache/flink/core/fs/RecoverableWriter.java

@@ -138,10 +138,8 @@ public interface RecoverableWriter {
  * recover from a (potential) failure. These can be temporary files that were written
  * to the filesystem or objects that were uploaded to S3.
  *
- * <p><b>NOTE:</b> This operation should not throw an exception if the resumable has already
- * been cleaned up and the resources have been freed. But the contract is that it will throw
- * an {@link UnsupportedOperationException} if it is called for a {@code RecoverableWriter}
- * whose {@link #requiresCleanupOfRecoverableState()} returns {@code false}.
+ * <p><b>NOTE:</b> This operation should not throw an exception, but return false if the cleanup did not
+ * happen for any reason.
  *
  * @param resumable The {@link ResumeRecoverable} whose state we want to clean-up.
  * @return {@code true} if the resources were successfully freed, {@code false} otherwise

flink-core/src/main/java/org/apache/flink/core/fs/local/LocalRecoverableWriter.java

@@ -77,7 +77,7 @@ public boolean requiresCleanupOfRecoverableState() {
 
  @Override
  public boolean cleanupRecoverableState(ResumeRecoverable resumable) throws IOException {
- throw new UnsupportedOperationException();
+ return false;
  }
 
  @Override

flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableWriter.java

@@ -95,7 +95,7 @@ public boolean requiresCleanupOfRecoverableState() {
 
  @Override
  public boolean cleanupRecoverableState(ResumeRecoverable resumable) throws IOException {
- throw new UnsupportedOperationException();
+ return false;
  }
 
  @Override

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

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http:https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.functions.sink.filesystem;
+
+/**
+ * An abstract writer for the currently open part file in a specific {@link Bucket}.
+ * @param <IN> the element type.
+ * @param <BucketID> the bucket id type.
+ */
+public abstract class AbstractPartFileWriter<IN, BucketID> implements InProgressFileWriter<IN, BucketID> {
+
+ private final BucketID bucketID;
+
+ private final long creationTime;
+
+ private long lastUpdateTime;
+
+ public AbstractPartFileWriter(final BucketID bucketID, final long createTime) {
+ this.bucketID = bucketID;
+ this.creationTime = createTime;
+ this.lastUpdateTime = createTime;
+ }
+
+ @Override
+ public BucketID getBucketId() {
+ return bucketID;
+ }
+
+ @Override
+ public long getCreationTime() {
+ return creationTime;
+ }
+
+ @Override
+ public long getLastUpdateTime() {
+ return lastUpdateTime;
+ }
+
+ void markWrite(long now) {
+ this.lastUpdateTime = now;
+ }
+}