[GitHub] zentol closed pull request #6437: [FLINK-9976][streaming] Remove unnecessary generic parameters

[GitBox <gi...@apache.org>]

zentol closed pull request #6437: [FLINK-9976][streaming] Remove unnecessary generic parameters
URL: https://github.com/apache/flink/pull/6437
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/StreamingFileSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/StreamingFileSink.java
index 0ebcc4f4c05..7daefc84826 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/StreamingFileSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/StreamingFileSink.java
@@ -186,52 +186,56 @@ private StreamingFileSink(
 
 		private static final long serialVersionUID = 1L;
 
-		private long bucketCheckInterval = 60L * 1000L;
+		private final long bucketCheckInterval;
 
 		private final Path basePath;
 
 		private final Encoder<IN> encoder;
 
-		private Bucketer<IN, BucketID> bucketer;
+		private final Bucketer<IN, BucketID> bucketer;
 
-		private RollingPolicy<IN, BucketID> rollingPolicy;
+		private final RollingPolicy<IN, BucketID> rollingPolicy;
 
-		private BucketFactory<IN, BucketID> bucketFactory = new DefaultBucketFactory<>();
+		private final BucketFactory<IN, BucketID> bucketFactory;
 
 		RowFormatBuilder(Path basePath, Encoder<IN> encoder, Bucketer<IN, BucketID> bucketer) {
+			this(basePath, encoder, bucketer, DefaultRollingPolicy.create().build(), 60L * 1000L, new DefaultBucketFactory<>());
+		}
+
+		private RowFormatBuilder(
+				Path basePath,
+				Encoder<IN> encoder,
+				Bucketer<IN, BucketID> bucketer,
+				RollingPolicy<IN, BucketID> rollingPolicy,
+				long bucketCheckInterval,
+				BucketFactory<IN, BucketID> bucketFactory) {
 			this.basePath = Preconditions.checkNotNull(basePath);
 			this.encoder = Preconditions.checkNotNull(encoder);
 			this.bucketer = Preconditions.checkNotNull(bucketer);
-			this.rollingPolicy = DefaultRollingPolicy.create().build();
+			this.rollingPolicy = Preconditions.checkNotNull(rollingPolicy);
+			this.bucketCheckInterval = bucketCheckInterval;
+			this.bucketFactory = Preconditions.checkNotNull(bucketFactory);
 		}
 
 		public StreamingFileSink.RowFormatBuilder<IN, BucketID> withBucketCheckInterval(final long interval) {
-			this.bucketCheckInterval = interval;
-			return this;
+			return new RowFormatBuilder<>(basePath, encoder, bucketer, rollingPolicy, interval, bucketFactory);
 		}
 
 		public StreamingFileSink.RowFormatBuilder<IN, BucketID> withBucketer(final Bucketer<IN, BucketID> bucketer) {
-			this.bucketer = Preconditions.checkNotNull(bucketer);
-			return this;
+			return new RowFormatBuilder<>(basePath, encoder, Preconditions.checkNotNull(bucketer), rollingPolicy, bucketCheckInterval, bucketFactory);
 		}
 
 		public StreamingFileSink.RowFormatBuilder<IN, BucketID> withRollingPolicy(final RollingPolicy<IN, BucketID> policy) {
-			this.rollingPolicy = Preconditions.checkNotNull(policy);
-			return this;
+			return new RowFormatBuilder<>(basePath, encoder, bucketer, Preconditions.checkNotNull(policy), bucketCheckInterval, bucketFactory);
 		}
 
 		public <ID> StreamingFileSink.RowFormatBuilder<IN, ID> withBucketerAndPolicy(final Bucketer<IN, ID> bucketer, final RollingPolicy<IN, ID> policy) {
-			@SuppressWarnings("unchecked")
-			StreamingFileSink.RowFormatBuilder<IN, ID> reInterpreted = (StreamingFileSink.RowFormatBuilder<IN, ID>) this;
-			reInterpreted.bucketer = Preconditions.checkNotNull(bucketer);
-			reInterpreted.rollingPolicy = Preconditions.checkNotNull(policy);
-			return reInterpreted;
+			return new RowFormatBuilder<>(basePath, encoder, Preconditions.checkNotNull(bucketer), Preconditions.checkNotNull(policy), bucketCheckInterval, new DefaultBucketFactory<>());
 		}
 
 		@VisibleForTesting
 		StreamingFileSink.RowFormatBuilder<IN, BucketID> withBucketFactory(final BucketFactory<IN, BucketID> factory) {
-			this.bucketFactory = Preconditions.checkNotNull(factory);
-			return this;
+			return new RowFormatBuilder<>(basePath, encoder, bucketer, rollingPolicy, bucketCheckInterval, Preconditions.checkNotNull(factory));
 		}
 
 		/** Creates the actual sink. */
@@ -259,38 +263,44 @@ private StreamingFileSink(
 
 		private static final long serialVersionUID = 1L;
 
-		private long bucketCheckInterval = 60L * 1000L;
+		private final long bucketCheckInterval;
 
 		private final Path basePath;
 
 		private final BulkWriter.Factory<IN> writerFactory;
 
-		private Bucketer<IN, BucketID> bucketer;
+		private final Bucketer<IN, BucketID> bucketer;
 
-		private BucketFactory<IN, BucketID> bucketFactory = new DefaultBucketFactory<>();
+		private final BucketFactory<IN, BucketID> bucketFactory;
 
 		BulkFormatBuilder(Path basePath, BulkWriter.Factory<IN> writerFactory, Bucketer<IN, BucketID> bucketer) {
+			this(basePath, writerFactory, bucketer, 60L * 1000L, new DefaultBucketFactory<>());
+		}
+
+		private BulkFormatBuilder(
+				Path basePath,
+				BulkWriter.Factory<IN> writerFactory,
+				Bucketer<IN, BucketID> bucketer,
+				long bucketCheckInterval,
+				BucketFactory<IN, BucketID> bucketFactory) {
 			this.basePath = Preconditions.checkNotNull(basePath);
-			this.writerFactory = Preconditions.checkNotNull(writerFactory);
+			this.writerFactory = writerFactory;
 			this.bucketer = Preconditions.checkNotNull(bucketer);
+			this.bucketCheckInterval = bucketCheckInterval;
+			this.bucketFactory = Preconditions.checkNotNull(bucketFactory);
 		}
 
 		public StreamingFileSink.BulkFormatBuilder<IN, BucketID> withBucketCheckInterval(long interval) {
-			this.bucketCheckInterval = interval;
-			return this;
+			return new BulkFormatBuilder<>(basePath, writerFactory, bucketer, interval, bucketFactory);
 		}
 
 		public <ID> StreamingFileSink.BulkFormatBuilder<IN, ID> withBucketer(Bucketer<IN, ID> bucketer) {
-			@SuppressWarnings("unchecked")
-			StreamingFileSink.BulkFormatBuilder<IN, ID> reInterpreted = (StreamingFileSink.BulkFormatBuilder<IN, ID>) this;
-			reInterpreted.bucketer = Preconditions.checkNotNull(bucketer);
-			return reInterpreted;
+			return new BulkFormatBuilder<>(basePath, writerFactory, Preconditions.checkNotNull(bucketer), bucketCheckInterval, new DefaultBucketFactory<>());
 		}
 
 		@VisibleForTesting
 		StreamingFileSink.BulkFormatBuilder<IN, BucketID> withBucketFactory(final BucketFactory<IN, BucketID> factory) {
-			this.bucketFactory = Preconditions.checkNotNull(factory);
-			return this;
+			return new BulkFormatBuilder<>(basePath, writerFactory, bucketer, bucketCheckInterval, Preconditions.checkNotNull(factory));
 		}
 
 		/** Creates the actual sink. */


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services