[
https://issues.apache.org/jira/browse/FLINK-9976?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16563358#comment-16563358
]
ASF GitHub Bot commented on FLINK-9976:
---------------------------------------
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:
[email protected]
> Odd signatures for streaming file sink format builders
> ------------------------------------------------------
>
> Key: FLINK-9976
> URL: https://issues.apache.org/jira/browse/FLINK-9976
> Project: Flink
> Issue Type: Bug
> Components: Streaming Connectors
> Affects Versions: 1.6.0
> Reporter: Chesnay Schepler
> Assignee: Chesnay Schepler
> Priority: Major
> Labels: pull-request-available
> Fix For: 1.6.0
>
>
> There are 2 instances of apparently unnecessary generic parameters in the
> format builders for the {{StreamingFileSink}}.
> Both these methods have a generic parameter for the BucketID type, however
> the builder itself already has such a parameter. The methods use unchecked
> casts to make the types fit, so we should be able to modify the signature to
> use the builders parameter instead.
> {code}
> public static class RowFormatBuilder<IN, BucketID> extends
> StreamingFileSink.BucketsBuilder<IN, BucketID> {
> ...
> 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;
> }
> ...
> {code}
> {code}
> public static class BulkFormatBuilder<IN, BucketID> extends
> StreamingFileSink.BucketsBuilder<IN, BucketID> {
> ...
> 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;
> }
> ...
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
