gemini-code-assist[bot] commented on code in PR #37036:
URL: https://github.com/apache/beam/pull/37036#discussion_r2600611548
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java:
##########
@@ -187,29 +196,127 @@ public static String
getCurrentTransformName(FlinkStreamingTranslationContext co
// Transformation Implementations
//
--------------------------------------------------------------------------------------------
+ /** Common translation logic for unbounded sources. */
+ @SuppressWarnings("unchecked")
+ private static <T> void translateUnboundedSource(
+ UnboundedSource<T, ?> rawSource,
+ String transformName,
+ FlinkStreamingTranslationContext context) {
+
+ PCollection<T> output =
+ (PCollection<T>)
+
Iterables.getOnlyElement(context.getCurrentTransform().getOutputs().values());
+
+ DataStream<WindowedValue<T>> source;
+ DataStream<WindowedValue<ValueWithRecordId<T>>> nonDedupSource;
+ TypeInformation<WindowedValue<T>> outputTypeInfo =
context.getTypeInfo(output);
+
+ Coder<T> coder = output.getCoder();
+
+ TypeInformation<WindowedValue<ValueWithRecordId<T>>> withIdTypeInfo =
+ new CoderTypeInformation<>(
+ WindowedValues.getFullCoder(
+ ValueWithRecordId.ValueWithRecordIdCoder.of(coder),
+ output.getWindowingStrategy().getWindowFn().windowCoder()),
+ context.getPipelineOptions());
+
+ String fullName = getCurrentTransformName(context);
+ try {
+ int parallelism =
+ context.getExecutionEnvironment().getMaxParallelism() > 0
+ ? context.getExecutionEnvironment().getMaxParallelism()
+ : context.getExecutionEnvironment().getParallelism();
Review Comment:
This logic for determining the source parallelism is duplicated in
`translateBoundedSource` (lines 277-280). To improve maintainability and reduce
code duplication, consider extracting this into a private helper method.
For example, you could add:
```java
private static int getSourceParallelism(FlinkStreamingTranslationContext
context) {
int maxParallelism =
context.getExecutionEnvironment().getMaxParallelism();
return maxParallelism > 0 ? maxParallelism :
context.getExecutionEnvironment().getParallelism();
}
```
And then replace the duplicated blocks in both `translateUnboundedSource`
and `translateBoundedSource` with:
```java
int parallelism = getSourceParallelism(context);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
