gemini-code-assist[bot] commented on code in PR #39316: URL: https://github.com/apache/beam/pull/39316#discussion_r3575229725
########## runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/KeyCommitTooLargeException.java: ########## @@ -17,17 +17,30 @@ */ package org.apache.beam.runners.dataflow.worker.streaming; +import com.google.protobuf.TextFormat; Review Comment:  The Windmill protobuf classes are generated using Beam's vendored gRPC/protobuf dependencies. Importing the standard `com.google.protobuf.TextFormat` will cause a compilation error due to type mismatches with the vendored `ByteString` returned by `request.getKey()`. Please use the vendored `TextFormat` class instead. ```suggestion import org.apache.beam.vendor.grpc.v1p69p0.com.google.protobuf.TextFormat; ``` ########## runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/StreamingWorkScheduler.java: ########## @@ -145,6 +149,9 @@ public static StreamingWorkScheduler create( hotKeyLogger, sideInputStateFetcherFactory); + boolean hotKeyLoggingEnabled = + options.isHotKeyLoggingEnabled() || hasExperiment(options, "enable_hot_key_logging"); Review Comment:  Since `DataflowRunner` cannot be imported in the worker module, check the experiments list directly from `options` to determine if hot key logging is enabled. ```java boolean hotKeyLoggingEnabled = options.isHotKeyLoggingEnabled() || (options.getExperiments() != null && options.getExperiments().stream() .anyMatch("enable_hot_key_logging"::equalsIgnoreCase)); ``` ########## runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/StreamingWorkScheduler.java: ########## @@ -17,10 +17,13 @@ */ package org.apache.beam.runners.dataflow.worker.windmill.work.processing; -import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; +import static com.google.common.base.Preconditions.checkState; +import static org.apache.beam.runners.dataflow.DataflowRunner.hasExperiment; import com.google.api.services.dataflow.model.MapTask; import com.google.auto.value.AutoValue; +import com.google.common.collect.ImmutableList; +import com.google.protobuf.ByteString; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentMap; Review Comment:  To prevent dependency conflicts, the Dataflow worker uses vendored versions of Guava and Protobuf. Additionally, importing `DataflowRunner` from the runner module into the worker module introduces a circular dependency and will fail to compile. Please use the vendored versions of Guava and Protobuf, and remove the `DataflowRunner` import. ```suggestion import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; import com.google.api.services.dataflow.model.MapTask; import com.google.auto.value.AutoValue; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentMap; import org.apache.beam.vendor.grpc.v1p69p0.com.google.protobuf.ByteString; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; ``` ########## runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java: ########## @@ -58,6 +58,19 @@ import com.google.api.services.dataflow.model.WorkItemStatus; import com.google.api.services.dataflow.model.WriteInstruction; import com.google.auto.value.AutoValue; +import com.google.common.cache.CacheStats; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Iterables; +import com.google.common.collect.Lists; +import com.google.common.primitives.UnsignedLong; +import com.google.common.util.concurrent.ThreadFactoryBuilder; +import com.google.common.util.concurrent.Uninterruptibles; +import com.google.protobuf.ByteString; +import com.google.protobuf.TextFormat; +import io.grpc.Server; +import io.grpc.ServerBuilder; +import io.grpc.testing.GrpcCleanupRule; Review Comment:  Please use the vendored versions of Guava, Protobuf, and gRPC to ensure compatibility with the worker's internal dependencies and avoid compilation type mismatches. ```suggestion import org.apache.beam.vendor.grpc.v1p69p0.com.google.protobuf.ByteString; import org.apache.beam.vendor.grpc.v1p69p0.com.google.protobuf.TextFormat; import org.apache.beam.vendor.grpc.v1p69p0.io.grpc.Server; import org.apache.beam.vendor.grpc.v1p69p0.io.grpc.ServerBuilder; import org.apache.beam.vendor.grpc.v1p69p0.io.grpc.testing.GrpcCleanupRule; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.cache.CacheStats; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.UnsignedLong; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.ThreadFactoryBuilder; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.Uninterruptibles; ``` -- 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]
