From 2473d9c0e9fdf56a5941f7685338b56cba669823 Mon Sep 17 00:00:00 2001
From: xujiantop-crypto <265865031+xujiantop-crypto@users.noreply.github.com>
Date: Wed, 9 Sep 2026 13:24:11 +0800
Subject: [PATCH] [Java] Extract common timestamp output DoFn

Signed-off-by: xujiantop-crypto <265865031+xujiantop-crypto@users.noreply.github.com>
---
 .../apache/beam/sdk/transforms/Create.java    | 14 +++++---
 .../transforms/OutputWithTimestampDoFn.java   | 33 +++++++++++++++++++
 .../beam/sdk/transforms/WithTimestamps.java   | 13 +++++---
 3 files changed, 52 insertions(+), 8 deletions(-)
 create mode 100644 sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/OutputWithTimestampDoFn.java

diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Create.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Create.java
index 50e08d438f46..a8f89d4c2fee 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Create.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Create.java
@@ -754,10 +754,16 @@ private TimestampedValues(
       this.typeDescriptor = typeDescriptor;
     }
 
-    private static class ConvertTimestamps<T> extends DoFn<TimestampedValue<T>, T> {
-      @ProcessElement
-      public void processElement(@Element TimestampedValue<T> element, OutputReceiver<T> r) {
-        r.outputWithTimestamp(element.getValue(), element.getTimestamp());
+    private static class ConvertTimestamps<T>
+        extends OutputWithTimestampDoFn<TimestampedValue<T>, T> {
+      @Override
+      T getOutput(TimestampedValue<T> element) {
+        return element.getValue();
+      }
+
+      @Override
+      Instant getTimestamp(TimestampedValue<T> element) {
+        return element.getTimestamp();
       }
     }
   }
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/OutputWithTimestampDoFn.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/OutputWithTimestampDoFn.java
new file mode 100644
index 000000000000..9d7cbdc980a7
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/OutputWithTimestampDoFn.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import org.joda.time.Instant;
+
+/** A {@link DoFn} that extracts an output value and timestamp from each input element. */
+abstract class OutputWithTimestampDoFn<InputT, OutputT> extends DoFn<InputT, OutputT> {
+
+  @ProcessElement
+  public void processElement(@Element InputT element, OutputReceiver<OutputT> receiver) {
+    receiver.outputWithTimestamp(getOutput(element), getTimestamp(element));
+  }
+
+  abstract OutputT getOutput(InputT element);
+
+  abstract Instant getTimestamp(InputT element);
+}
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/WithTimestamps.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/WithTimestamps.java
index 31ff3e6eeb47..43fc7421003b 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/WithTimestamps.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/WithTimestamps.java
@@ -119,7 +119,7 @@ public PCollection<T> expand(PCollection<T> input) {
         "AddTimestamps", ParDo.of(new AddTimestampsDoFn<>(fn, allowedTimestampSkew)));
   }
 
-  private static class AddTimestampsDoFn<T> extends DoFn<T, T> {
+  private static class AddTimestampsDoFn<T> extends OutputWithTimestampDoFn<T, T> {
     private final SerializableFunction<T, Instant> fn;
     private final Duration allowedTimestampSkew;
 
@@ -128,12 +128,17 @@ public AddTimestampsDoFn(SerializableFunction<T, Instant> fn, Duration allowedTi
       this.allowedTimestampSkew = allowedTimestampSkew;
     }
 
-    @ProcessElement
-    public void processElement(@Element T element, OutputReceiver<T> r) {
+    @Override
+    T getOutput(T element) {
+      return element;
+    }
+
+    @Override
+    Instant getTimestamp(T element) {
       Instant timestamp = fn.apply(element);
       checkNotNull(
           timestamp, "Timestamps for WithTimestamps cannot be null. Timestamp provided by %s.", fn);
-      r.outputWithTimestamp(element, timestamp);
+      return timestamp;
     }
 
     @Override