[FLINK-29530] CommittableMessageTypeInfo#toString may force client to have S3 credentials

• Type: Technical Debt
• Status: Open
• Priority: Critical
• Resolution: Unresolved
• Affects Version/s: 1.15.0
• Fix Version/s: 1.20.0
• Component/s: Connectors / FileSystem
• Assignee: Unassigned
• Reporter: Chesnay Schepler
• Created: 06/Oct/22 14:59
• Updated: 11/Mar/24 12:43

Description

The toString implementation calls the committable serializer factory, which in the case of the FileSink loads a filesystem and thus may require credentials.
As a result something as basic as logging the type info can force the client to require S3 credentials.

It shouldn't have such side-effects......

Stacktrace that a user provided for another issue:

Exception in thread "main" org.apache.flink.util.FlinkRuntimeException: Could not create committable serializer.
	at org.apache.flink.connector.file.sink.FileSink.getCommittableSerializer(FileSink.java:180)
	at org.apache.flink.streaming.api.connector.sink2.CommittableMessageTypeInfo.toString(CommittableMessageTypeInfo.java:120)
	at java.base/java.lang.String.valueOf(String.java:2951)
	at java.base/java.lang.StringBuilder.append(StringBuilder.java:172)
	at org.apache.flink.api.dag.Transformation.toString(Transformation.java:556)
	at java.base/java.lang.String.valueOf(String.java:2951)
	at java.base/java.lang.StringBuilder.append(StringBuilder.java:172)
	at org.apache.flink.streaming.api.graph.StreamGraphGenerator.transform(StreamGraphGenerator.java:506)
	at org.apache.flink.streaming.api.graph.StreamGraphGenerator.access$200(StreamGraphGenerator.java:131)
	at org.apache.flink.streaming.api.graph.StreamGraphGenerator$ContextImpl.transform(StreamGraphGenerator.java:933)
	at java.base/java.lang.Iterable.forEach(Iterable.java:75)
	at org.apache.flink.streaming.runtime.translators.SinkTransformationTranslator$SinkExpander.expand(SinkTransformationTranslator.java:157)
	at org.apache.flink.streaming.runtime.translators.SinkTransformationTranslator$SinkExpander.access$000(SinkTransformationTranslator.java:99)
	at org.apache.flink.streaming.runtime.translators.SinkTransformationTranslator.translateInternal(SinkTransformationTranslator.java:89)
	at org.apache.flink.streaming.runtime.translators.SinkTransformationTranslator.translateForStreaming(SinkTransformationTranslator.java:77)
	at org.apache.flink.streaming.runtime.translators.SinkTransformationTranslator.translateForStreaming(SinkTransformationTranslator.java:61)
	at org.apache.flink.streaming.api.graph.StreamGraphGenerator.translate(StreamGraphGenerator.java:825)
	at org.apache.flink.streaming.api.graph.StreamGraphGenerator.transform(StreamGraphGenerator.java:555)
	at org.apache.flink.streaming.api.graph.StreamGraphGenerator.generate(StreamGraphGenerator.java:316)
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getStreamGraph(StreamExecutionEnvironment.java:2135)
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getStreamGraph(StreamExecutionEnvironment.java:2121)
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1967)