[FLINK-9976][streaming] Remove unnecessary generic parameters

[zentol]

What is the purpose of the change

This PR removes unnecessary generic parameter from format builder methods of the StreamingFileSink.
These are inconsistent with other existing methods that use the already defined bucket ID type of the builder, and as of right now would always lead to ClassCastExceptions if one tried to use a different type.

[aljoscha]

Isn't this meant for the case where you want to change the BucketID, so you actually need the reinterpretation?

[zentol]

but the bucket ID type is already defined in the constructor. you can't just change that.

[aljoscha]

I think the Builders are done wrong. Instead of reusing the builder, the builder should be immutable and each call should return a new builder that has fields from the old builder and the new value that the user wants to set. This way we don't need to reinterpret. WDYT?

[zentol]

I think that would be a nicer approach. The fact that this compiles and runs without error is rather disgusting:

final RowFormatBuilder<String, String> original = forRowFormat(new Path("test"), (Encoder<String>) (element, stream) -> stream.write(0));
final RowFormatBuilder<String, Integer> casted = original
	.withBucketerAndPolicy(new Bucketer<String, Integer>() {
		@Override
		public Integer getBucketId(String element, Context context) {
			return 0;
		}

		@Override
		public SimpleVersionedSerializer<Integer> getSerializer() {
			return null;
		}
	}, new RollingPolicy<String, Integer>() {
		@Override
		public boolean shouldRollOnCheckpoint(PartFileInfo<Integer> partFileState) throws IOException {
			return false;
		}

		@Override
		public boolean shouldRollOnEvent(PartFileInfo<Integer> partFileState, String element) throws IOException {
			return false;
		}

		@Override
		public boolean shouldRollOnProcessingTime(PartFileInfo<Integer> partFileState, long currentTime) throws IOException {
			return false;
		}
	});

if (!original.equals(casted)) {
	throw new RuntimeException();
}

[zentol]

@aljoscha I've reworked the builders. This introduced a small gotcha for tests; if you set the bucketFactory, and afterwards change the type the factory is overridden with a default factory.

[aljoscha]

👍 that's better now

[yanghua]

Shall we also change the PR's title?

[kl0u]

I agree that the code looks nicer now, but essentially the only thing changed is that in the:

if (!original.equals(casted)) {
	throw new RuntimeException();
}

now we throw the RuntimeException, right?

[kl0u]

This constructor can be private. Actually also the other constructor could be, as it is called by the enclosing class, but if it were to move the builder to a different file, then it would have a problem. So I would recommend to make the new constructor private.

[kl0u]

I would also call this constructor from the other one, just to have a single point of assignment of variables.

[kl0u]

Same as above, this constructor can be private and call this constructor from the previous one.

[zentol]

yes, the functionality is the same, and in my example the test would now fail.