Naireen commented on code in PR #34275:
URL: https://github.com/apache/beam/pull/34275#discussion_r2006325029
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java:
##########
@@ -454,122 +454,125 @@ public ProcessContinuation processElement(
final Stopwatch sw = Stopwatch.createStarted();
while (true) {
- rawRecords = poll(consumer, kafkaSourceDescriptor.getTopicPartition());
- // When there are no records available for the current TopicPartition,
self-checkpoint
- // and move to process the next element.
- if (rawRecords.isEmpty()) {
- if (!topicPartitionExists(
- kafkaSourceDescriptor.getTopicPartition(),
- consumer.partitionsFor(kafkaSourceDescriptor.getTopic()))) {
- return ProcessContinuation.stop();
- }
- if (timestampPolicy != null) {
- updateWatermarkManually(timestampPolicy, watermarkEstimator,
tracker);
- }
- return ProcessContinuation.resume();
- }
- for (ConsumerRecord<byte[], byte[]> rawRecord : rawRecords) {
- // If the Kafka consumer returns a record with an offset that is
already processed
- // the record can be safely skipped. This is needed because there is
a possibility
- // that the seek() above fails to move the offset to the desired
position. In which
- // case poll() would return records that are already cnsumed.
- if (rawRecord.offset() < startOffset) {
- // If the start offset is not reached even after skipping the
records for 10 seconds
- // then the processing is stopped with a backoff to give the Kakfa
server some time
- // catch up.
- if (sw.elapsed().getSeconds() > 10L) {
- LOG.error(
- "The expected offset ({}) was not reached even after"
- + " skipping consumed records for 10 seconds. The offset
we could"
- + " reach was {}. The processing of this bundle will be
attempted"
- + " at a later time.",
- expectedOffset,
- rawRecord.offset());
- return ProcessContinuation.resume()
-
.withResumeDelay(org.joda.time.Duration.standardSeconds(10L));
- }
- skippedRecords++;
- continue;
- }
- if (skippedRecords > 0L) {
- LOG.warn(
- "{} records were skipped due to seek returning an"
- + " earlier position than requested position of {}",
- skippedRecords,
- expectedOffset);
- skippedRecords = 0L;
- }
- if (!tracker.tryClaim(rawRecord.offset())) {
- return ProcessContinuation.stop();
- }
- try {
- KafkaRecord<K, V> kafkaRecord =
- new KafkaRecord<>(
- rawRecord.topic(),
- rawRecord.partition(),
- rawRecord.offset(),
- ConsumerSpEL.getRecordTimestamp(rawRecord),
- ConsumerSpEL.getRecordTimestampType(rawRecord),
- ConsumerSpEL.hasHeaders() ? rawRecord.headers() : null,
- ConsumerSpEL.deserializeKey(keyDeserializerInstance,
rawRecord),
- ConsumerSpEL.deserializeValue(valueDeserializerInstance,
rawRecord));
- int recordSize =
- (rawRecord.key() == null ? 0 : rawRecord.key().length)
- + (rawRecord.value() == null ? 0 :
rawRecord.value().length);
- avgRecordSizeCache
- .getUnchecked(kafkaSourceDescriptor)
- .update(recordSize, rawRecord.offset() - expectedOffset);
- rawSizes.update(recordSize);
- expectedOffset = rawRecord.offset() + 1;
- Instant outputTimestamp;
- // The outputTimestamp and watermark will be computed by
timestampPolicy, where the
- // WatermarkEstimator should be a manual one.
- if (timestampPolicy != null) {
- TimestampPolicyContext context =
- updateWatermarkManually(timestampPolicy, watermarkEstimator,
tracker);
- outputTimestamp = timestampPolicy.getTimestampForRecord(context,
kafkaRecord);
- } else {
- Preconditions.checkStateNotNull(this.extractOutputTimestampFn);
- outputTimestamp = extractOutputTimestampFn.apply(kafkaRecord);
+ KafkaMetrics kafkaMetrics = KafkaSinkMetrics.kafkaMetrics();
Review Comment:
So the way the container is implemented is that once its flushed, we dont
reuse it. I'm less usre of the context behind this decision, I dont see why we
cant change it, as I agree it would make it more efficient, I'll leave that for
a subsequent PR.
--
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: github-unsubscr...@beam.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
