gemini-code-assist[bot] commented on code in PR #39285:
URL: https://github.com/apache/beam/pull/39285#discussion_r3560776001
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java:
##########
@@ -687,7 +634,7 @@ public ProcessContinuation processElement(
final long estimatedBacklogBytes =
(long)
- (BigDecimal.valueOf(latestOffsetEstimator.estimate())
+ (BigDecimal.valueOf(latestOffsetEstimator.get())
.subtract(BigDecimal.valueOf(expectedOffset),
MathContext.DECIMAL128)
.doubleValue()
* avgRecordSize.get());
Review Comment:
If `latestOffsetEstimator` is still `Long.MIN_VALUE` (e.g., if the first
poll has not completed or failed to update the estimator), calculating the
backlog using `Long.MIN_VALUE` will result in a huge negative backlog value.
This can severely disrupt the runner's autoscaling and progress tracking. A
defensive check should be added to default the backlog to `0` if the estimate
is unknown.
```java
final long estimate = latestOffsetEstimator.get();
final long estimatedBacklogBytes =
estimate == Long.MIN_VALUE
? 0L
: (long)
(BigDecimal.valueOf(estimate)
.subtract(BigDecimal.valueOf(expectedOffset),
MathContext.DECIMAL128)
.doubleValue()
* avgRecordSize.get());
```
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java:
##########
@@ -510,9 +449,10 @@ public UnsplittableRestrictionTracker<OffsetRange, Long>
restrictionTracker(
// OffsetEstimators are cached for each topic-partition because they hold
a stateful connection,
// so we want to minimize the amount of connections that we start and
track with Kafka. Another
// point is that it has a memoized backlog, and this should make that more
reusable estimations.
+ final AtomicLong latestOffsetEstimator =
+ latestOffsetEstimatorCache.getUnchecked(kafkaSourceDescriptor);
return new UnsplittableRestrictionTracker<>(
- new GrowableOffsetRangeTracker(
- restriction.getFrom(),
latestOffsetEstimatorCache.getUnchecked(kafkaSourceDescriptor)));
+ new GrowableOffsetRangeTracker(restriction.getFrom(),
latestOffsetEstimator::get));
Review Comment:
Since `latestOffsetEstimator` is initialized to `Long.MIN_VALUE`, the
`GrowableOffsetRangeTracker` will receive `Long.MIN_VALUE` as the end offset
estimate before the first poll completes. This causes `getProgress()` to cap
the end offset to the current position and report `0` remaining work, which can
incorrectly signal to the runner that the partition is fully processed.
Returning `Long.MAX_VALUE` when the estimate is unknown (i.e.,
`Long.MIN_VALUE`) represents an infinite/unknown end offset, which is much
safer and more appropriate for an unbounded source.
```java
final AtomicLong latestOffsetEstimator =
latestOffsetEstimatorCache.getUnchecked(kafkaSourceDescriptor);
return new UnsplittableRestrictionTracker<>(
new GrowableOffsetRangeTracker(
restriction.getFrom(),
() -> {
long estimate = latestOffsetEstimator.get();
return estimate == Long.MIN_VALUE ? Long.MAX_VALUE : estimate;
}));
```
##########
runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/utils/EmbeddedKafkaCluster.java:
##########
@@ -114,15 +114,20 @@ public void startup() {
properties.setProperty("offsets.topic.replication.factor", "1");
properties.setProperty("log.flush.interval.messages", String.valueOf(1));
- KafkaServerStartable broker = startBroker(properties);
+ KafkaServer broker = startBroker(properties);
brokers.add(broker);
logDirs.add(logDir);
}
}
- private static KafkaServerStartable startBroker(Properties props) {
- KafkaServerStartable server = new KafkaServerStartable(new
KafkaConfig(props));
+ private static KafkaServer startBroker(Properties props) {
+ KafkaServer server =
+ new KafkaServer(
+ new KafkaConfig(props),
+ KafkaServer.$lessinit$greater$default$2(),
+ KafkaServer.$lessinit$greater$default$3(),
+ KafkaServer.$lessinit$greater$default$4());
Review Comment:
Using Scala synthetic methods like `$lessinit$greater$default$X` for default
arguments is extremely fragile and can break if the Scala version or compiler
internals change. It is much more robust to pass explicit, standard arguments
such as `Time.SYSTEM`, `Option.empty()`, and `Nil$.MODULE$`.
```suggestion
KafkaServer server =
new KafkaServer(
new KafkaConfig(props),
org.apache.kafka.common.utils.Time.SYSTEM,
scala.Option.empty(),
scala.collection.immutable.Nil$.MODULE$);
```
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java:
##########
@@ -580,6 +519,14 @@ public ProcessContinuation processElement(
// Fetch the next records.
final ConsumerRecords<byte[], byte[]> rawRecords =
consumer.poll(remainingTimeout);
final Duration elapsed = pollTimer.elapsed();
+ try {
+ final long position = consumer.position(topicPartition);
+ consumer
+ .currentLag(topicPartition)
+ .ifPresent(lag -> latestOffsetEstimator.lazySet(position + lag));
+ } catch (KafkaException e) {
+ }
Review Comment:
Empty catch blocks violate coding standards (such as the Google Java Style
Guide) and can hide important errors. It is highly recommended to at least log
the exception at a debug level to aid in troubleshooting.
```suggestion
} catch (KafkaException e) {
LOG.debug("Failed to update end offset estimator using local
consumer state", e);
}
```
<details>
<summary>References</summary>
1. Empty catch blocks are allowed only when the exception is expected and
documented, or when it is ignored on purpose with a comment explaining why.
<sup>([link](http://console.cloud.google.com/gemini-code-assist/agents-tools))</sup>
</details>
##########
sdks/java/testing/kafka-service/src/test/java/org/apache/beam/sdk/testing/kafka/LocalKafka.java:
##########
@@ -20,18 +20,23 @@
import java.nio.file.Files;
import java.util.Properties;
import kafka.server.KafkaConfig;
-import kafka.server.KafkaServerStartable;
+import kafka.server.KafkaServer;
public class LocalKafka {
- private final KafkaServerStartable server;
+ private final KafkaServer server;
LocalKafka(int kafkaPort, int zookeeperPort) throws Exception {
Properties kafkaProperties = new Properties();
kafkaProperties.setProperty("port", String.valueOf(kafkaPort));
kafkaProperties.setProperty("zookeeper.connect",
String.format("localhost:%s", zookeeperPort));
kafkaProperties.setProperty("offsets.topic.replication.factor", "1");
kafkaProperties.setProperty("log.dir",
Files.createTempDirectory("kafka-log-").toString());
- server = new KafkaServerStartable(KafkaConfig.fromProps(kafkaProperties));
+ server =
+ new KafkaServer(
+ KafkaConfig.fromProps(kafkaProperties),
+ KafkaServer.$lessinit$greater$default$2(),
+ KafkaServer.$lessinit$greater$default$3(),
+ KafkaServer.$lessinit$greater$default$4());
Review Comment:
Using Scala synthetic methods like `$lessinit$greater$default$X` for default
arguments is extremely fragile and can break if the Scala version or compiler
internals change. It is much more robust to pass explicit, standard arguments
such as `Time.SYSTEM`, `Option.empty()`, and `Nil$.MODULE$`.
```suggestion
server =
new KafkaServer(
KafkaConfig.fromProps(kafkaProperties),
org.apache.kafka.common.utils.Time.SYSTEM,
scala.Option.empty(),
scala.collection.immutable.Nil$.MODULE$);
```
--
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]
