You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "lalitx17 (via GitHub)" <gi...@apache.org> on 2026/05/19 05:06:08 UTC
[PR] feat: Add LogElements transform to Java SDK [beam]
lalitx17 opened a new pull request, #38533:
URL: https://github.com/apache/beam/pull/38533
**Please** add a meaningful description for your change here
## What does this PR do?
Added a Java SDK LogElements transform for logging each element of a PCollection while passing the elements through unchanged.
Supports SLF4J log levels plus optional prefix, timestamp, window, and pane info logging, mirroring the Python SDK’s LogElements convenience transform.
addresses #38528
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #38533:
URL: https://github.com/apache/beam/pull/38533#discussion_r3356658983
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/LogElements.java:
##########
@@ -0,0 +1,234 @@
+/*
+ * 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 java.util.Objects;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.event.Level;
+
+/**
+ * {@link PTransform} for logging elements of a {@link PCollection}.
+ *
+ * <p>Each element is logged and then emitted unchanged.
+ *
+ * <pre>{@code
+ * PCollection<String> words = ...;
+ * PCollection<String> loggedWords = words.apply(LogElements.info().withPrefix("word: "));
+ * }</pre>
+ *
+ * @param <T> the element type of the input {@link PCollection}
+ */
+public class LogElements<T> extends PTransform<PCollection<T>, PCollection<T>> {
+ private static final Logger LOG = LoggerFactory.getLogger(LogElements.class);
+
+ private final Level level;
+ private final String prefix;
+ private final boolean withTimestamp;
+ private final boolean withWindow;
+ private final boolean withPaneInfo;
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#TRACE}. */
+ public static <T> LogElements<T> trace() {
+ return of(Level.TRACE);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#DEBUG}. */
+ public static <T> LogElements<T> debug() {
+ return of(Level.DEBUG);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#INFO}. */
+ public static <T> LogElements<T> info() {
+ return of(Level.INFO);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#WARN}. */
+ public static <T> LogElements<T> warn() {
+ return of(Level.WARN);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#ERROR}. */
+ public static <T> LogElements<T> error() {
+ return of(Level.ERROR);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at the given level. */
+ public static <T> LogElements<T> of(Level level) {
+ return new LogElements<>(level, "", false, false, false);
+ }
+
+ private LogElements(
+ Level level, String prefix, boolean withTimestamp, boolean withWindow, boolean withPaneInfo) {
+ this.level = Objects.requireNonNull(level, "level");
+ this.prefix = Objects.requireNonNull(prefix, "prefix");
+ this.withTimestamp = withTimestamp;
+ this.withWindow = withWindow;
+ this.withPaneInfo = withPaneInfo;
+ }
+
+ /** Returns a new {@link LogElements} transform with the given prefix before each element. */
+ public LogElements<T> withPrefix(String prefix) {
+ return new LogElements<>(level, prefix, withTimestamp, withWindow, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's timestamp. */
+ public LogElements<T> withTimestamp() {
+ return new LogElements<>(level, prefix, true, withWindow, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's window. */
+ public LogElements<T> withWindow() {
+ return new LogElements<>(level, prefix, withTimestamp, true, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's pane info. */
+ public LogElements<T> withPaneInfo() {
+ return new LogElements<>(level, prefix, withTimestamp, withWindow, true);
+ }
+
+ @Override
+ public PCollection<T> expand(PCollection<T> input) {
+ return input.apply("Log", ParDo.of(new LoggingFn<>(this)));
+ }
+
+ @Override
+ public void populateDisplayData(DisplayData.Builder builder) {
+ super.populateDisplayData(builder);
+ builder
+ .add(DisplayData.item("level", level.name()).withLabel("Log Level"))
+ .addIfNotDefault(DisplayData.item("prefix", prefix).withLabel("Prefix"), "")
+ .addIfNotDefault(
+ DisplayData.item("withTimestamp", withTimestamp).withLabel("Log Timestamp"), false)
+ .addIfNotDefault(DisplayData.item("withWindow", withWindow).withLabel("Log Window"), false)
+ .addIfNotDefault(
+ DisplayData.item("withPaneInfo", withPaneInfo).withLabel("Log Pane Info"), false);
+ }
+
+ static String formatForLogging(
+ @Nullable Object element,
+ String prefix,
+ boolean withTimestamp,
+ boolean withWindow,
+ boolean withPaneInfo,
+ Instant timestamp,
+ BoundedWindow window,
+ PaneInfo paneInfo) {
+ StringBuilder builder = new StringBuilder(prefix).append(element);
+ if (withTimestamp) {
+ builder.append(", timestamp=").append(timestamp);
+ }
+ if (withWindow) {
+ builder.append(", window=").append(window);
+ }
+ if (withPaneInfo) {
+ builder.append(", paneInfo=").append(paneInfo);
+ }
+ return builder.toString();
+ }
+
+ @VisibleForTesting
+ static void log(Level level, String message) {
+ switch (level) {
+ case TRACE:
+ LOG.trace("{}", message);
+ break;
+ case DEBUG:
+ LOG.debug("{}", message);
+ break;
+ case INFO:
+ LOG.info("{}", message);
+ break;
+ case WARN:
+ LOG.warn("{}", message);
+ break;
+ case ERROR:
+ LOG.error("{}", message);
+ break;
+ default:
+ throw unsupportedLogLevel(level);
+ }
+ }
+
+ private static boolean isLoggingEnabled(Level level) {
+ switch (level) {
+ case TRACE:
+ return LOG.isTraceEnabled();
+ case DEBUG:
+ return LOG.isDebugEnabled();
+ case INFO:
+ return LOG.isInfoEnabled();
+ case WARN:
+ return LOG.isWarnEnabled();
+ case ERROR:
+ return LOG.isErrorEnabled();
+ default:
+ throw unsupportedLogLevel(level);
+ }
+ }
+
+ private static IllegalArgumentException unsupportedLogLevel(Level level) {
+ return new IllegalArgumentException("Unsupported log level: " + level);
+ }
+
+ private static class LoggingFn<T> extends DoFn<T, T> {
+ private final Level level;
+ private final String prefix;
+ private final boolean withTimestamp;
+ private final boolean withWindow;
+ private final boolean withPaneInfo;
+
+ private LoggingFn(LogElements<T> transform) {
+ this.level = transform.level;
+ this.prefix = transform.prefix;
+ this.withTimestamp = transform.withTimestamp;
+ this.withWindow = transform.withWindow;
+ this.withPaneInfo = transform.withPaneInfo;
+ }
+
+ @ProcessElement
+ public void processElement(
+ @Element T element,
+ @DoFn.Timestamp Instant timestamp,
+ BoundedWindow window,
+ PaneInfo paneInfo,
Review Comment:
Beam DoFn parameter injection for window/pane typically requires the parameter annotations (e.g., `@DoFn.WindowParam` for `BoundedWindow` and `@DoFn.PaneInfoParam` for `PaneInfo`). Without these, DoFn signature validation may fail (or these values may not be injected as intended). Consider adding the appropriate `DoFn` param annotations to make the injection explicit and robust.
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/LogElements.java:
##########
@@ -0,0 +1,234 @@
+/*
+ * 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 java.util.Objects;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.event.Level;
+
+/**
+ * {@link PTransform} for logging elements of a {@link PCollection}.
+ *
+ * <p>Each element is logged and then emitted unchanged.
+ *
+ * <pre>{@code
+ * PCollection<String> words = ...;
+ * PCollection<String> loggedWords = words.apply(LogElements.info().withPrefix("word: "));
+ * }</pre>
+ *
+ * @param <T> the element type of the input {@link PCollection}
+ */
+public class LogElements<T> extends PTransform<PCollection<T>, PCollection<T>> {
+ private static final Logger LOG = LoggerFactory.getLogger(LogElements.class);
+
+ private final Level level;
+ private final String prefix;
+ private final boolean withTimestamp;
+ private final boolean withWindow;
+ private final boolean withPaneInfo;
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#TRACE}. */
+ public static <T> LogElements<T> trace() {
+ return of(Level.TRACE);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#DEBUG}. */
+ public static <T> LogElements<T> debug() {
+ return of(Level.DEBUG);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#INFO}. */
+ public static <T> LogElements<T> info() {
+ return of(Level.INFO);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#WARN}. */
+ public static <T> LogElements<T> warn() {
+ return of(Level.WARN);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#ERROR}. */
+ public static <T> LogElements<T> error() {
+ return of(Level.ERROR);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at the given level. */
+ public static <T> LogElements<T> of(Level level) {
+ return new LogElements<>(level, "", false, false, false);
+ }
+
+ private LogElements(
+ Level level, String prefix, boolean withTimestamp, boolean withWindow, boolean withPaneInfo) {
+ this.level = Objects.requireNonNull(level, "level");
+ this.prefix = Objects.requireNonNull(prefix, "prefix");
+ this.withTimestamp = withTimestamp;
+ this.withWindow = withWindow;
+ this.withPaneInfo = withPaneInfo;
+ }
+
+ /** Returns a new {@link LogElements} transform with the given prefix before each element. */
+ public LogElements<T> withPrefix(String prefix) {
+ return new LogElements<>(level, prefix, withTimestamp, withWindow, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's timestamp. */
+ public LogElements<T> withTimestamp() {
+ return new LogElements<>(level, prefix, true, withWindow, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's window. */
+ public LogElements<T> withWindow() {
+ return new LogElements<>(level, prefix, withTimestamp, true, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's pane info. */
+ public LogElements<T> withPaneInfo() {
+ return new LogElements<>(level, prefix, withTimestamp, withWindow, true);
+ }
+
+ @Override
+ public PCollection<T> expand(PCollection<T> input) {
+ return input.apply("Log", ParDo.of(new LoggingFn<>(this)));
+ }
+
+ @Override
+ public void populateDisplayData(DisplayData.Builder builder) {
+ super.populateDisplayData(builder);
+ builder
+ .add(DisplayData.item("level", level.name()).withLabel("Log Level"))
+ .addIfNotDefault(DisplayData.item("prefix", prefix).withLabel("Prefix"), "")
+ .addIfNotDefault(
+ DisplayData.item("withTimestamp", withTimestamp).withLabel("Log Timestamp"), false)
+ .addIfNotDefault(DisplayData.item("withWindow", withWindow).withLabel("Log Window"), false)
+ .addIfNotDefault(
+ DisplayData.item("withPaneInfo", withPaneInfo).withLabel("Log Pane Info"), false);
+ }
+
+ static String formatForLogging(
+ @Nullable Object element,
+ String prefix,
+ boolean withTimestamp,
+ boolean withWindow,
+ boolean withPaneInfo,
+ Instant timestamp,
+ BoundedWindow window,
+ PaneInfo paneInfo) {
+ StringBuilder builder = new StringBuilder(prefix).append(element);
+ if (withTimestamp) {
+ builder.append(", timestamp=").append(timestamp);
+ }
+ if (withWindow) {
+ builder.append(", window=").append(window);
+ }
+ if (withPaneInfo) {
+ builder.append(", paneInfo=").append(paneInfo);
+ }
+ return builder.toString();
+ }
+
+ @VisibleForTesting
+ static void log(Level level, String message) {
+ switch (level) {
+ case TRACE:
+ LOG.trace("{}", message);
+ break;
+ case DEBUG:
+ LOG.debug("{}", message);
+ break;
+ case INFO:
+ LOG.info("{}", message);
+ break;
+ case WARN:
+ LOG.warn("{}", message);
+ break;
+ case ERROR:
+ LOG.error("{}", message);
+ break;
+ default:
+ throw unsupportedLogLevel(level);
+ }
+ }
+
+ private static boolean isLoggingEnabled(Level level) {
+ switch (level) {
+ case TRACE:
+ return LOG.isTraceEnabled();
+ case DEBUG:
+ return LOG.isDebugEnabled();
+ case INFO:
+ return LOG.isInfoEnabled();
+ case WARN:
+ return LOG.isWarnEnabled();
+ case ERROR:
+ return LOG.isErrorEnabled();
+ default:
+ throw unsupportedLogLevel(level);
+ }
+ }
+
+ private static IllegalArgumentException unsupportedLogLevel(Level level) {
+ return new IllegalArgumentException("Unsupported log level: " + level);
+ }
+
+ private static class LoggingFn<T> extends DoFn<T, T> {
+ private final Level level;
+ private final String prefix;
+ private final boolean withTimestamp;
+ private final boolean withWindow;
+ private final boolean withPaneInfo;
+
+ private LoggingFn(LogElements<T> transform) {
+ this.level = transform.level;
+ this.prefix = transform.prefix;
+ this.withTimestamp = transform.withTimestamp;
+ this.withWindow = transform.withWindow;
+ this.withPaneInfo = transform.withPaneInfo;
+ }
+
+ @ProcessElement
+ public void processElement(
+ @Element T element,
+ @DoFn.Timestamp Instant timestamp,
+ BoundedWindow window,
+ PaneInfo paneInfo,
+ OutputReceiver<T> receiver) {
Review Comment:
`@ProcessElement`, `@Element`, and `OutputReceiver` are referenced as simple names but (in Beam) these are nested types on `DoFn` and typically must be imported (e.g., `import org.apache.beam.sdk.transforms.DoFn.ProcessElement;`, `DoFn.Element`, `DoFn.OutputReceiver`) or referenced as `@DoFn.ProcessElement`, `@DoFn.Element`, and `DoFn.OutputReceiver<T>`. As written, this will fail to compile if those simple names are not otherwise defined in the package.
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "lalitx17 (via GitHub)" <gi...@apache.org>.
lalitx17 commented on PR #38533:
URL: https://github.com/apache/beam/pull/38533#issuecomment-4490459543
assign set of reviewers
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on code in PR #38533:
URL: https://github.com/apache/beam/pull/38533#discussion_r3276266225
##########
sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/LogElementsTest.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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 static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+
+import java.util.Arrays;
+import java.util.List;
+import org.apache.beam.sdk.testing.NeedsRunner;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.values.PCollection;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.event.Level;
+
+/** Tests for {@link LogElements}. */
+@RunWith(JUnit4.class)
+public class LogElementsTest {
Review Comment:
Is there a way to verify that the transform indeed logs elements at the correct level?
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on code in PR #38533:
URL: https://github.com/apache/beam/pull/38533#discussion_r3263802049
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/LogElements.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * 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 java.util.Objects;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.values.PCollection;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.event.Level;
+
+/**
+ * {@link PTransform} for logging elements of a {@link PCollection}.
+ *
+ * <p>Each element is logged and then emitted unchanged.
+ *
+ * <pre>{@code
+ * PCollection<String> words = ...;
+ * PCollection<String> loggedWords = words.apply(LogElements.info().withPrefix("word: "));
+ * }</pre>
+ *
+ * @param <T> the element type of the input {@link PCollection}
+ */
+public class LogElements<T> extends PTransform<PCollection<T>, PCollection<T>> {
+ private static final Logger LOG = LoggerFactory.getLogger(LogElements.class);
+
+ private final Level level;
+ private final String prefix;
+ private final boolean withTimestamp;
+ private final boolean withWindow;
+ private final boolean withPaneInfo;
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#TRACE}. */
+ public static <T> LogElements<T> trace() {
+ return of(Level.TRACE);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#DEBUG}. */
+ public static <T> LogElements<T> debug() {
+ return of(Level.DEBUG);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#INFO}. */
+ public static <T> LogElements<T> info() {
+ return of(Level.INFO);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#WARN}. */
+ public static <T> LogElements<T> warn() {
+ return of(Level.WARN);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#ERROR}. */
+ public static <T> LogElements<T> error() {
+ return of(Level.ERROR);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at the given level. */
+ public static <T> LogElements<T> of(Level level) {
+ return new LogElements<>(level, "", false, false, false);
+ }
+
+ private LogElements(
+ Level level, String prefix, boolean withTimestamp, boolean withWindow, boolean withPaneInfo) {
+ this.level = Objects.requireNonNull(level, "level");
+ this.prefix = Objects.requireNonNull(prefix, "prefix");
+ this.withTimestamp = withTimestamp;
+ this.withWindow = withWindow;
+ this.withPaneInfo = withPaneInfo;
+ }
+
+ /** Returns a new {@link LogElements} transform with the given prefix before each element. */
+ public LogElements<T> withPrefix(String prefix) {
+ return new LogElements<>(level, prefix, withTimestamp, withWindow, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's timestamp. */
+ public LogElements<T> withTimestamp() {
+ return new LogElements<>(level, prefix, true, withWindow, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's window. */
+ public LogElements<T> withWindow() {
+ return new LogElements<>(level, prefix, withTimestamp, true, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's pane info. */
+ public LogElements<T> withPaneInfo() {
+ return new LogElements<>(level, prefix, withTimestamp, withWindow, true);
+ }
+
+ @Override
+ public PCollection<T> expand(PCollection<T> input) {
+ return input.apply("Log", ParDo.of(new LoggingFn<>(this))).setCoder(input.getCoder());
+ }
+
+ @Override
+ public void populateDisplayData(DisplayData.Builder builder) {
+ super.populateDisplayData(builder);
+ builder
+ .add(DisplayData.item("level", level.name()).withLabel("Log Level"))
+ .addIfNotDefault(DisplayData.item("prefix", prefix).withLabel("Prefix"), "")
+ .addIfNotDefault(
+ DisplayData.item("withTimestamp", withTimestamp).withLabel("Log Timestamp"), false)
+ .addIfNotDefault(DisplayData.item("withWindow", withWindow).withLabel("Log Window"), false)
+ .addIfNotDefault(
+ DisplayData.item("withPaneInfo", withPaneInfo).withLabel("Log Pane Info"), false);
+ }
+
+ static String formatForLogging(
+ @Nullable Object element,
+ String prefix,
+ boolean withTimestamp,
+ boolean withWindow,
+ boolean withPaneInfo,
+ Instant timestamp,
+ BoundedWindow window,
+ PaneInfo paneInfo) {
+ StringBuilder builder = new StringBuilder(prefix).append(element);
+ if (withTimestamp) {
+ builder.append(", timestamp=").append(timestamp);
+ }
+ if (withWindow) {
+ builder.append(", window=").append(window);
+ }
+ if (withPaneInfo) {
+ builder.append(", paneInfo=").append(paneInfo);
+ }
+ return builder.toString();
+ }
+
+ private static void log(Level level, String message) {
+ switch (level) {
+ case TRACE:
+ LOG.trace("{}", message);
+ break;
+ case DEBUG:
+ LOG.debug("{}", message);
+ break;
+ case INFO:
+ LOG.info("{}", message);
+ break;
+ case WARN:
+ LOG.warn("{}", message);
+ break;
+ case ERROR:
+ default:
+ LOG.error("{}", message);
+ }
+ }
+
+ private static class LoggingFn<T> extends DoFn<T, T> {
+ private final Level level;
+ private final String prefix;
+ private final boolean withTimestamp;
+ private final boolean withWindow;
+ private final boolean withPaneInfo;
+
+ private LoggingFn(LogElements<T> transform) {
+ this.level = transform.level;
+ this.prefix = transform.prefix;
+ this.withTimestamp = transform.withTimestamp;
+ this.withWindow = transform.withWindow;
+ this.withPaneInfo = transform.withPaneInfo;
+ }
+
+ @ProcessElement
+ public void processElement(
+ @Element T element,
+ @DoFn.Timestamp Instant timestamp,
+ BoundedWindow window,
+ PaneInfo paneInfo,
+ OutputReceiver<T> receiver) {
+ log(
+ level,
+ formatForLogging(
+ element,
+ prefix,
+ withTimestamp,
+ withWindow,
+ withPaneInfo,
+ timestamp,
+ window,
+ paneInfo));
+ receiver.output(element);
+ }
Review Comment:
The `formatForLogging` method is called for every element before checking if the log level is enabled. This involves string concatenation and `toString()` calls (potentially on large objects, windows, or timestamps) which can be very expensive in high-throughput pipelines. You should check if the log level is enabled before constructing the log message.
```java
if (isLoggingEnabled(level)) {
log(
level,
formatForLogging(
element,
prefix,
withTimestamp,
withWindow,
withPaneInfo,
timestamp,
window,
paneInfo));
}
receiver.output(element);
}
private boolean isLoggingEnabled(Level level) {
switch (level) {
case TRACE:
return LOG.isTraceEnabled();
case DEBUG:
return LOG.isDebugEnabled();
case INFO:
return LOG.isInfoEnabled();
case WARN:
return LOG.isWarnEnabled();
case ERROR:
default:
return LOG.isErrorEnabled();
}
}
```
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/LogElements.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * 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 java.util.Objects;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.values.PCollection;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.event.Level;
+
+/**
+ * {@link PTransform} for logging elements of a {@link PCollection}.
+ *
+ * <p>Each element is logged and then emitted unchanged.
+ *
+ * <pre>{@code
+ * PCollection<String> words = ...;
+ * PCollection<String> loggedWords = words.apply(LogElements.info().withPrefix("word: "));
+ * }</pre>
+ *
+ * @param <T> the element type of the input {@link PCollection}
+ */
+public class LogElements<T> extends PTransform<PCollection<T>, PCollection<T>> {
+ private static final Logger LOG = LoggerFactory.getLogger(LogElements.class);
+
+ private final Level level;
+ private final String prefix;
+ private final boolean withTimestamp;
+ private final boolean withWindow;
+ private final boolean withPaneInfo;
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#TRACE}. */
+ public static <T> LogElements<T> trace() {
+ return of(Level.TRACE);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#DEBUG}. */
+ public static <T> LogElements<T> debug() {
+ return of(Level.DEBUG);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#INFO}. */
+ public static <T> LogElements<T> info() {
+ return of(Level.INFO);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#WARN}. */
+ public static <T> LogElements<T> warn() {
+ return of(Level.WARN);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#ERROR}. */
+ public static <T> LogElements<T> error() {
+ return of(Level.ERROR);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at the given level. */
+ public static <T> LogElements<T> of(Level level) {
+ return new LogElements<>(level, "", false, false, false);
+ }
+
+ private LogElements(
+ Level level, String prefix, boolean withTimestamp, boolean withWindow, boolean withPaneInfo) {
+ this.level = Objects.requireNonNull(level, "level");
+ this.prefix = Objects.requireNonNull(prefix, "prefix");
+ this.withTimestamp = withTimestamp;
+ this.withWindow = withWindow;
+ this.withPaneInfo = withPaneInfo;
+ }
+
+ /** Returns a new {@link LogElements} transform with the given prefix before each element. */
+ public LogElements<T> withPrefix(String prefix) {
+ return new LogElements<>(level, prefix, withTimestamp, withWindow, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's timestamp. */
+ public LogElements<T> withTimestamp() {
+ return new LogElements<>(level, prefix, true, withWindow, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's window. */
+ public LogElements<T> withWindow() {
+ return new LogElements<>(level, prefix, withTimestamp, true, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's pane info. */
+ public LogElements<T> withPaneInfo() {
+ return new LogElements<>(level, prefix, withTimestamp, withWindow, true);
+ }
+
+ @Override
+ public PCollection<T> expand(PCollection<T> input) {
+ return input.apply("Log", ParDo.of(new LoggingFn<>(this))).setCoder(input.getCoder());
Review Comment:
Calling `setCoder(input.getCoder())` is generally redundant for a `ParDo` that returns the same type as its input, as Beam automatically propagates the coder. Additionally, `input.getCoder()` can throw an `IllegalStateException` if the coder hasn't been set yet and cannot be inferred at this stage of pipeline construction. It is safer to let the SDK handle coder propagation.
```suggestion
return input.apply("Log", ParDo.of(new LoggingFn<>(this)));
```
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "lalitx17 (via GitHub)" <gi...@apache.org>.
lalitx17 commented on PR #38533:
URL: https://github.com/apache/beam/pull/38533#issuecomment-4622072618
> What are the next steps here ? Thanks!
Thanks for the ping @chamikaramj! I’ll move this forward by avoiding the runner-dependent test.
Given that `ExpectedLogs` only works reliably when SLF4J is bridged to JUL, and DirectRunner uses `slf4j-simple`, I don’t think changing the project-wide SLF4J binding or capturing stderr is worth the blast radius/flakiness.
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "lalitx17 (via GitHub)" <gi...@apache.org>.
lalitx17 commented on PR #38533:
URL: https://github.com/apache/beam/pull/38533#issuecomment-4490723398
R: @ahmedabu98 please take a look. Thanks.
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38533:
URL: https://github.com/apache/beam/pull/38533#issuecomment-4490741019
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment `assign set of reviewers`
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "lalitx17 (via GitHub)" <gi...@apache.org>.
lalitx17 commented on PR #38533:
URL: https://github.com/apache/beam/pull/38533#issuecomment-4512409763
so the issue is that the DirectRunner uses slf4j-simple which outputs to stderr, which ExpectedLogs can't capture(only works with JUL).
also slf4j-simple has trace and debug disabled by default.
solutions like changing slf4j-simple to slf4j-jdk14 has high blast radius.
the next most obvious path is to capture stderr(trace and debug still won't appear) but can be flaky. (bad idea)
Lmk if you have better idea.
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 merged PR #38533:
URL: https://github.com/apache/beam/pull/38533
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "lalitx17 (via GitHub)" <gi...@apache.org>.
lalitx17 commented on PR #38533:
URL: https://github.com/apache/beam/pull/38533#issuecomment-4489095665
> .free-issue
I'm almost done.
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "ashwin-sampathkumar (via GitHub)" <gi...@apache.org>.
ashwin-sampathkumar commented on PR #38533:
URL: https://github.com/apache/beam/pull/38533#issuecomment-4488912399
.free-issue
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38533:
URL: https://github.com/apache/beam/pull/38533#issuecomment-4490489546
Reviewers are already assigned to this PR: @chamikaramj
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "lalitx17 (via GitHub)" <gi...@apache.org>.
lalitx17 commented on PR #38533:
URL: https://github.com/apache/beam/pull/38533#issuecomment-4490483388
retest this please
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "lalitx17 (via GitHub)" <gi...@apache.org>.
lalitx17 commented on PR #38533:
URL: https://github.com/apache/beam/pull/38533#issuecomment-4510424864
really appreciate the speedy review.
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "lalitx17 (via GitHub)" <gi...@apache.org>.
lalitx17 commented on code in PR #38533:
URL: https://github.com/apache/beam/pull/38533#discussion_r3276785667
##########
sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/LogElementsTest.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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 static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+
+import java.util.Arrays;
+import java.util.List;
+import org.apache.beam.sdk.testing.NeedsRunner;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.values.PCollection;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.event.Level;
+
+/** Tests for {@link LogElements}. */
+@RunWith(JUnit4.class)
+public class LogElementsTest {
Review Comment:
yeah. so I put a commit where **ExpectedLogs** methods like _verifyTrace_ verifies transform logs elements at expected level.
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on PR #38533:
URL: https://github.com/apache/beam/pull/38533#issuecomment-4511102179
Looks like the new test failed:
```
2026-05-21T17:04:06.3026397Z LogElementsTest > testLogElementsLogsAtConfiguredLevels FAILED
2026-05-21T17:04:06.3057465Z java.lang.AssertionError at LogElementsTest.java:88
```
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38533:
URL: https://github.com/apache/beam/pull/38533#issuecomment-4485751209
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment `assign set of reviewers`
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "lalitx17 (via GitHub)" <gi...@apache.org>.
lalitx17 commented on PR #38533:
URL: https://github.com/apache/beam/pull/38533#issuecomment-4490456147
the fail checks seems unrelated to my commit. so
assign set of reviewers
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #38533:
URL: https://github.com/apache/beam/pull/38533#discussion_r3263794927
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/LogElements.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * 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 java.util.Objects;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.values.PCollection;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.event.Level;
+
+/**
+ * {@link PTransform} for logging elements of a {@link PCollection}.
+ *
+ * <p>Each element is logged and then emitted unchanged.
+ *
+ * <pre>{@code
+ * PCollection<String> words = ...;
+ * PCollection<String> loggedWords = words.apply(LogElements.info().withPrefix("word: "));
+ * }</pre>
+ *
+ * @param <T> the element type of the input {@link PCollection}
+ */
+public class LogElements<T> extends PTransform<PCollection<T>, PCollection<T>> {
+ private static final Logger LOG = LoggerFactory.getLogger(LogElements.class);
+
+ private final Level level;
+ private final String prefix;
+ private final boolean withTimestamp;
+ private final boolean withWindow;
+ private final boolean withPaneInfo;
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#TRACE}. */
+ public static <T> LogElements<T> trace() {
+ return of(Level.TRACE);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#DEBUG}. */
+ public static <T> LogElements<T> debug() {
+ return of(Level.DEBUG);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#INFO}. */
+ public static <T> LogElements<T> info() {
+ return of(Level.INFO);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#WARN}. */
+ public static <T> LogElements<T> warn() {
+ return of(Level.WARN);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#ERROR}. */
+ public static <T> LogElements<T> error() {
+ return of(Level.ERROR);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at the given level. */
+ public static <T> LogElements<T> of(Level level) {
+ return new LogElements<>(level, "", false, false, false);
+ }
+
+ private LogElements(
+ Level level, String prefix, boolean withTimestamp, boolean withWindow, boolean withPaneInfo) {
+ this.level = Objects.requireNonNull(level, "level");
+ this.prefix = Objects.requireNonNull(prefix, "prefix");
+ this.withTimestamp = withTimestamp;
+ this.withWindow = withWindow;
+ this.withPaneInfo = withPaneInfo;
+ }
+
+ /** Returns a new {@link LogElements} transform with the given prefix before each element. */
+ public LogElements<T> withPrefix(String prefix) {
+ return new LogElements<>(level, prefix, withTimestamp, withWindow, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's timestamp. */
+ public LogElements<T> withTimestamp() {
+ return new LogElements<>(level, prefix, true, withWindow, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's window. */
+ public LogElements<T> withWindow() {
+ return new LogElements<>(level, prefix, withTimestamp, true, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's pane info. */
+ public LogElements<T> withPaneInfo() {
+ return new LogElements<>(level, prefix, withTimestamp, withWindow, true);
+ }
+
+ @Override
+ public PCollection<T> expand(PCollection<T> input) {
+ return input.apply("Log", ParDo.of(new LoggingFn<>(this))).setCoder(input.getCoder());
+ }
+
+ @Override
+ public void populateDisplayData(DisplayData.Builder builder) {
+ super.populateDisplayData(builder);
+ builder
+ .add(DisplayData.item("level", level.name()).withLabel("Log Level"))
+ .addIfNotDefault(DisplayData.item("prefix", prefix).withLabel("Prefix"), "")
+ .addIfNotDefault(
+ DisplayData.item("withTimestamp", withTimestamp).withLabel("Log Timestamp"), false)
+ .addIfNotDefault(DisplayData.item("withWindow", withWindow).withLabel("Log Window"), false)
+ .addIfNotDefault(
+ DisplayData.item("withPaneInfo", withPaneInfo).withLabel("Log Pane Info"), false);
+ }
+
+ static String formatForLogging(
+ @Nullable Object element,
+ String prefix,
+ boolean withTimestamp,
+ boolean withWindow,
+ boolean withPaneInfo,
+ Instant timestamp,
+ BoundedWindow window,
+ PaneInfo paneInfo) {
+ StringBuilder builder = new StringBuilder(prefix).append(element);
+ if (withTimestamp) {
+ builder.append(", timestamp=").append(timestamp);
+ }
+ if (withWindow) {
+ builder.append(", window=").append(window);
+ }
+ if (withPaneInfo) {
+ builder.append(", paneInfo=").append(paneInfo);
+ }
+ return builder.toString();
+ }
+
+ private static void log(Level level, String message) {
+ switch (level) {
+ case TRACE:
+ LOG.trace("{}", message);
+ break;
+ case DEBUG:
+ LOG.debug("{}", message);
+ break;
+ case INFO:
+ LOG.info("{}", message);
+ break;
+ case WARN:
+ LOG.warn("{}", message);
+ break;
+ case ERROR:
+ default:
+ LOG.error("{}", message);
Review Comment:
Folding the `default` branch into `ERROR` will silently log at ERROR level if SLF4J ever introduces a new `Level` value (or if a caller passes an unexpected value). Consider handling `ERROR` explicitly and throwing `IllegalArgumentException` (or logging a warning) in `default`, so unsupported levels are surfaced rather than masked as errors.
##########
sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/LogElementsTest.java:
##########
@@ -0,0 +1,86 @@
+/*
+ * 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 static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+
+import java.util.Arrays;
+import java.util.List;
+import org.apache.beam.sdk.testing.NeedsRunner;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.values.PCollection;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.event.Level;
+
+/** Tests for {@link LogElements}. */
+@RunWith(JUnit4.class)
+public class LogElementsTest {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ @Test
+ @Category(NeedsRunner.class)
+ public void testLogElementsPreservesElements() {
+ List<String> elements = Arrays.asList("a", "b", "c");
+
+ PCollection<String> output =
+ pipeline.apply(Create.of(elements)).apply(LogElements.<String>info());
+
+ PAssert.that(output).containsInAnyOrder(elements);
+ pipeline.run();
+ }
+
+ @Test
+ public void testFormatForLoggingIncludesConfiguredMetadata() {
+ Instant timestamp = new Instant(0);
+ IntervalWindow window = new IntervalWindow(timestamp, Duration.standardMinutes(1));
+
+ String message =
+ LogElements.formatForLogging(
+ "a", "row: ", true, true, true, timestamp, window, PaneInfo.NO_FIRING);
+
+ assertThat(message, containsString("row: a"));
+ assertThat(message, containsString("timestamp=1970-01-01T00:00:00.000Z"));
+ assertThat(
+ message, containsString("window=[1970-01-01T00:00:00.000Z..1970-01-01T00:01:00.000Z)"));
+ assertThat(message, containsString("paneInfo=PaneInfo.NO_FIRING"));
+ }
+
+ @Test
+ public void testDisplayData() {
+ DisplayData displayData =
+ DisplayData.from(
+ LogElements.of(Level.WARN).withPrefix("row: ").withTimestamp().withWindow());
+
+ assertThat(displayData, hasDisplayItem("level", "WARN"));
+ assertThat(displayData, hasDisplayItem("prefix", "row: "));
+ assertThat(displayData, hasDisplayItem("withTimestamp", true));
+ assertThat(displayData, hasDisplayItem("withWindow", true));
+ }
Review Comment:
The tests cover element pass-through, formatting, and display data, but no test verifies that `LoggingFn.processElement` actually emits a log record at the configured level (e.g., via an SLF4J test appender or `ExpectedLogs`). Since the level-to-logger dispatch in `log(Level, String)` is the core behavior of this transform, adding a test that asserts a record was logged at each supported `Level` would catch regressions in the `switch` statement. Also consider extending `testDisplayData` to assert `withPaneInfo` (the only builder option not exercised here).
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #38533:
URL: https://github.com/apache/beam/pull/38533#issuecomment-4484594953
## Summary of Changes
Hello, I'm Gemini Code Assist[^1]! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
This pull request adds a convenient logging utility to the Apache Beam Java SDK, mirroring the existing functionality in the Python SDK. It enables developers to easily inspect PCollection elements during pipeline execution without altering the data flow, which is particularly useful for debugging and monitoring data processing pipelines.
### Highlights
* **New LogElements Transform**: Introduced a new LogElements PTransform in the Java SDK that allows logging elements of a PCollection while passing them through unchanged.
* **Flexible Logging Configuration**: Supports SLF4J log levels and provides optional configuration for including prefixes, timestamps, windowing, and pane information.
* **Testing**: Added comprehensive unit tests to verify element preservation, metadata formatting, and display data population.
<details>
<summary><b>New Features</b></summary>
🧠 You can now enable **Memory** (public preview) to help **Gemini Code Assist** learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. **Click [here](https://codeassist.google/code-review/login) to enable Memory in your admin console.**
</details>
<details>
<summary><b>Using Gemini Code Assist</b></summary>
<br>
The full guide for Gemini Code Assist can be found on our [documentation page](https://developers.google.com/gemini-code-assist/docs/review-github-code), here are some quick tips.
<b>Invoking Gemini</b>
You can request assistance from Gemini at any point by creating a comment using either `/gemini <command>` or `@gemini-code-assist <command>`. Below is a summary of the supported commands on the current page.
Feature | Command | Description
--- | --- | ---
Code Review | `/gemini review` | Performs a code review for the current pull request in its current state.
Pull Request Summary | `/gemini summary` | Provides a summary of the current pull request in its current state.
Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help | `/gemini help` | Displays a list of available commands.
<b>Customization</b>
To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a `.gemini/` folder in the base of the repository. Detailed instructions can be found [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github).
<b>Limitations & Feedback</b>
Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with :thumbsup: and :thumbsdown: on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up [here](https://google.qualtrics.com/jfe/form/SV_2cyuGuTWsEw84yG).
</details>
[^1]: Review the [Privacy Notices](https://policies.google.com/privacy), [Generative AI Prohibited Use Policy](https://policies.google.com/terms/generative-ai/use-policy), [Terms of Service](https://policies.google.com/terms), and learn how to configure Gemini Code Assist in GitHub [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github). Gemini can make mistakes, so double check it and [use code with caution](https://support.google.com/legal/answer/13505487).
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on code in PR #38533:
URL: https://github.com/apache/beam/pull/38533#discussion_r3282769767
##########
sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/LogElementsTest.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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 static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+
+import java.util.Arrays;
+import java.util.List;
+import org.apache.beam.sdk.testing.NeedsRunner;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.values.PCollection;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.event.Level;
+
+/** Tests for {@link LogElements}. */
+@RunWith(JUnit4.class)
+public class LogElementsTest {
Review Comment:
Amazing, thanks for doing that!
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "lalitx17 (via GitHub)" <gi...@apache.org>.
lalitx17 commented on code in PR #38533:
URL: https://github.com/apache/beam/pull/38533#discussion_r3357022919
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/LogElements.java:
##########
@@ -0,0 +1,234 @@
+/*
+ * 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 java.util.Objects;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.event.Level;
+
+/**
+ * {@link PTransform} for logging elements of a {@link PCollection}.
+ *
+ * <p>Each element is logged and then emitted unchanged.
+ *
+ * <pre>{@code
+ * PCollection<String> words = ...;
+ * PCollection<String> loggedWords = words.apply(LogElements.info().withPrefix("word: "));
+ * }</pre>
+ *
+ * @param <T> the element type of the input {@link PCollection}
+ */
+public class LogElements<T> extends PTransform<PCollection<T>, PCollection<T>> {
+ private static final Logger LOG = LoggerFactory.getLogger(LogElements.class);
+
+ private final Level level;
+ private final String prefix;
+ private final boolean withTimestamp;
+ private final boolean withWindow;
+ private final boolean withPaneInfo;
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#TRACE}. */
+ public static <T> LogElements<T> trace() {
+ return of(Level.TRACE);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#DEBUG}. */
+ public static <T> LogElements<T> debug() {
+ return of(Level.DEBUG);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#INFO}. */
+ public static <T> LogElements<T> info() {
+ return of(Level.INFO);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#WARN}. */
+ public static <T> LogElements<T> warn() {
+ return of(Level.WARN);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#ERROR}. */
+ public static <T> LogElements<T> error() {
+ return of(Level.ERROR);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at the given level. */
+ public static <T> LogElements<T> of(Level level) {
+ return new LogElements<>(level, "", false, false, false);
+ }
+
+ private LogElements(
+ Level level, String prefix, boolean withTimestamp, boolean withWindow, boolean withPaneInfo) {
+ this.level = Objects.requireNonNull(level, "level");
+ this.prefix = Objects.requireNonNull(prefix, "prefix");
+ this.withTimestamp = withTimestamp;
+ this.withWindow = withWindow;
+ this.withPaneInfo = withPaneInfo;
+ }
+
+ /** Returns a new {@link LogElements} transform with the given prefix before each element. */
+ public LogElements<T> withPrefix(String prefix) {
+ return new LogElements<>(level, prefix, withTimestamp, withWindow, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's timestamp. */
+ public LogElements<T> withTimestamp() {
+ return new LogElements<>(level, prefix, true, withWindow, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's window. */
+ public LogElements<T> withWindow() {
+ return new LogElements<>(level, prefix, withTimestamp, true, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's pane info. */
+ public LogElements<T> withPaneInfo() {
+ return new LogElements<>(level, prefix, withTimestamp, withWindow, true);
+ }
+
+ @Override
+ public PCollection<T> expand(PCollection<T> input) {
+ return input.apply("Log", ParDo.of(new LoggingFn<>(this)));
+ }
+
+ @Override
+ public void populateDisplayData(DisplayData.Builder builder) {
+ super.populateDisplayData(builder);
+ builder
+ .add(DisplayData.item("level", level.name()).withLabel("Log Level"))
+ .addIfNotDefault(DisplayData.item("prefix", prefix).withLabel("Prefix"), "")
+ .addIfNotDefault(
+ DisplayData.item("withTimestamp", withTimestamp).withLabel("Log Timestamp"), false)
+ .addIfNotDefault(DisplayData.item("withWindow", withWindow).withLabel("Log Window"), false)
+ .addIfNotDefault(
+ DisplayData.item("withPaneInfo", withPaneInfo).withLabel("Log Pane Info"), false);
+ }
+
+ static String formatForLogging(
+ @Nullable Object element,
+ String prefix,
+ boolean withTimestamp,
+ boolean withWindow,
+ boolean withPaneInfo,
+ Instant timestamp,
+ BoundedWindow window,
+ PaneInfo paneInfo) {
+ StringBuilder builder = new StringBuilder(prefix).append(element);
+ if (withTimestamp) {
+ builder.append(", timestamp=").append(timestamp);
+ }
+ if (withWindow) {
+ builder.append(", window=").append(window);
+ }
+ if (withPaneInfo) {
+ builder.append(", paneInfo=").append(paneInfo);
+ }
+ return builder.toString();
+ }
+
+ @VisibleForTesting
+ static void log(Level level, String message) {
+ switch (level) {
+ case TRACE:
+ LOG.trace("{}", message);
+ break;
+ case DEBUG:
+ LOG.debug("{}", message);
+ break;
+ case INFO:
+ LOG.info("{}", message);
+ break;
+ case WARN:
+ LOG.warn("{}", message);
+ break;
+ case ERROR:
+ LOG.error("{}", message);
+ break;
+ default:
+ throw unsupportedLogLevel(level);
+ }
+ }
+
+ private static boolean isLoggingEnabled(Level level) {
+ switch (level) {
+ case TRACE:
+ return LOG.isTraceEnabled();
+ case DEBUG:
+ return LOG.isDebugEnabled();
+ case INFO:
+ return LOG.isInfoEnabled();
+ case WARN:
+ return LOG.isWarnEnabled();
+ case ERROR:
+ return LOG.isErrorEnabled();
+ default:
+ throw unsupportedLogLevel(level);
+ }
+ }
+
+ private static IllegalArgumentException unsupportedLogLevel(Level level) {
+ return new IllegalArgumentException("Unsupported log level: " + level);
+ }
+
+ private static class LoggingFn<T> extends DoFn<T, T> {
+ private final Level level;
+ private final String prefix;
+ private final boolean withTimestamp;
+ private final boolean withWindow;
+ private final boolean withPaneInfo;
+
+ private LoggingFn(LogElements<T> transform) {
+ this.level = transform.level;
+ this.prefix = transform.prefix;
+ this.withTimestamp = transform.withTimestamp;
+ this.withWindow = transform.withWindow;
+ this.withPaneInfo = transform.withPaneInfo;
+ }
+
+ @ProcessElement
+ public void processElement(
+ @Element T element,
+ @DoFn.Timestamp Instant timestamp,
+ BoundedWindow window,
+ PaneInfo paneInfo,
+ OutputReceiver<T> receiver) {
Review Comment:
false positive
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38533:
URL: https://github.com/apache/beam/pull/38533#issuecomment-4490471300
Assigning reviewers:
R: @chamikaramj for label java.
Note: If you would like to opt out of this review, comment `assign to next reviewer`.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "lalitx17 (via GitHub)" <gi...@apache.org>.
lalitx17 commented on code in PR #38533:
URL: https://github.com/apache/beam/pull/38533#discussion_r3276785667
##########
sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/LogElementsTest.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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 static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+
+import java.util.Arrays;
+import java.util.List;
+import org.apache.beam.sdk.testing.NeedsRunner;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.values.PCollection;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.event.Level;
+
+/** Tests for {@link LogElements}. */
+@RunWith(JUnit4.class)
+public class LogElementsTest {
Review Comment:
yeah. so I put a commit where **ExpectedLogs** methods like _verifyTrace_ verifies transform logs elements at expected/configured level.
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on PR #38533:
URL: https://github.com/apache/beam/pull/38533#issuecomment-4510387431
Will merge when tests go green
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "chamikaramj (via GitHub)" <gi...@apache.org>.
chamikaramj commented on PR #38533:
URL: https://github.com/apache/beam/pull/38533#issuecomment-4597286108
What are the next steps here ? Thanks!
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add LogElements transform to Java SDK [beam]
Posted by "lalitx17 (via GitHub)" <gi...@apache.org>.
lalitx17 commented on code in PR #38533:
URL: https://github.com/apache/beam/pull/38533#discussion_r3357036051
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/LogElements.java:
##########
@@ -0,0 +1,234 @@
+/*
+ * 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 java.util.Objects;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.event.Level;
+
+/**
+ * {@link PTransform} for logging elements of a {@link PCollection}.
+ *
+ * <p>Each element is logged and then emitted unchanged.
+ *
+ * <pre>{@code
+ * PCollection<String> words = ...;
+ * PCollection<String> loggedWords = words.apply(LogElements.info().withPrefix("word: "));
+ * }</pre>
+ *
+ * @param <T> the element type of the input {@link PCollection}
+ */
+public class LogElements<T> extends PTransform<PCollection<T>, PCollection<T>> {
+ private static final Logger LOG = LoggerFactory.getLogger(LogElements.class);
+
+ private final Level level;
+ private final String prefix;
+ private final boolean withTimestamp;
+ private final boolean withWindow;
+ private final boolean withPaneInfo;
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#TRACE}. */
+ public static <T> LogElements<T> trace() {
+ return of(Level.TRACE);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#DEBUG}. */
+ public static <T> LogElements<T> debug() {
+ return of(Level.DEBUG);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#INFO}. */
+ public static <T> LogElements<T> info() {
+ return of(Level.INFO);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#WARN}. */
+ public static <T> LogElements<T> warn() {
+ return of(Level.WARN);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at {@link Level#ERROR}. */
+ public static <T> LogElements<T> error() {
+ return of(Level.ERROR);
+ }
+
+ /** Returns a {@link LogElements} transform that logs elements at the given level. */
+ public static <T> LogElements<T> of(Level level) {
+ return new LogElements<>(level, "", false, false, false);
+ }
+
+ private LogElements(
+ Level level, String prefix, boolean withTimestamp, boolean withWindow, boolean withPaneInfo) {
+ this.level = Objects.requireNonNull(level, "level");
+ this.prefix = Objects.requireNonNull(prefix, "prefix");
+ this.withTimestamp = withTimestamp;
+ this.withWindow = withWindow;
+ this.withPaneInfo = withPaneInfo;
+ }
+
+ /** Returns a new {@link LogElements} transform with the given prefix before each element. */
+ public LogElements<T> withPrefix(String prefix) {
+ return new LogElements<>(level, prefix, withTimestamp, withWindow, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's timestamp. */
+ public LogElements<T> withTimestamp() {
+ return new LogElements<>(level, prefix, true, withWindow, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's window. */
+ public LogElements<T> withWindow() {
+ return new LogElements<>(level, prefix, withTimestamp, true, withPaneInfo);
+ }
+
+ /** Returns a new {@link LogElements} transform that logs each element's pane info. */
+ public LogElements<T> withPaneInfo() {
+ return new LogElements<>(level, prefix, withTimestamp, withWindow, true);
+ }
+
+ @Override
+ public PCollection<T> expand(PCollection<T> input) {
+ return input.apply("Log", ParDo.of(new LoggingFn<>(this)));
+ }
+
+ @Override
+ public void populateDisplayData(DisplayData.Builder builder) {
+ super.populateDisplayData(builder);
+ builder
+ .add(DisplayData.item("level", level.name()).withLabel("Log Level"))
+ .addIfNotDefault(DisplayData.item("prefix", prefix).withLabel("Prefix"), "")
+ .addIfNotDefault(
+ DisplayData.item("withTimestamp", withTimestamp).withLabel("Log Timestamp"), false)
+ .addIfNotDefault(DisplayData.item("withWindow", withWindow).withLabel("Log Window"), false)
+ .addIfNotDefault(
+ DisplayData.item("withPaneInfo", withPaneInfo).withLabel("Log Pane Info"), false);
+ }
+
+ static String formatForLogging(
+ @Nullable Object element,
+ String prefix,
+ boolean withTimestamp,
+ boolean withWindow,
+ boolean withPaneInfo,
+ Instant timestamp,
+ BoundedWindow window,
+ PaneInfo paneInfo) {
+ StringBuilder builder = new StringBuilder(prefix).append(element);
+ if (withTimestamp) {
+ builder.append(", timestamp=").append(timestamp);
+ }
+ if (withWindow) {
+ builder.append(", window=").append(window);
+ }
+ if (withPaneInfo) {
+ builder.append(", paneInfo=").append(paneInfo);
+ }
+ return builder.toString();
+ }
+
+ @VisibleForTesting
+ static void log(Level level, String message) {
+ switch (level) {
+ case TRACE:
+ LOG.trace("{}", message);
+ break;
+ case DEBUG:
+ LOG.debug("{}", message);
+ break;
+ case INFO:
+ LOG.info("{}", message);
+ break;
+ case WARN:
+ LOG.warn("{}", message);
+ break;
+ case ERROR:
+ LOG.error("{}", message);
+ break;
+ default:
+ throw unsupportedLogLevel(level);
+ }
+ }
+
+ private static boolean isLoggingEnabled(Level level) {
+ switch (level) {
+ case TRACE:
+ return LOG.isTraceEnabled();
+ case DEBUG:
+ return LOG.isDebugEnabled();
+ case INFO:
+ return LOG.isInfoEnabled();
+ case WARN:
+ return LOG.isWarnEnabled();
+ case ERROR:
+ return LOG.isErrorEnabled();
+ default:
+ throw unsupportedLogLevel(level);
+ }
+ }
+
+ private static IllegalArgumentException unsupportedLogLevel(Level level) {
+ return new IllegalArgumentException("Unsupported log level: " + level);
+ }
+
+ private static class LoggingFn<T> extends DoFn<T, T> {
+ private final Level level;
+ private final String prefix;
+ private final boolean withTimestamp;
+ private final boolean withWindow;
+ private final boolean withPaneInfo;
+
+ private LoggingFn(LogElements<T> transform) {
+ this.level = transform.level;
+ this.prefix = transform.prefix;
+ this.withTimestamp = transform.withTimestamp;
+ this.withWindow = transform.withWindow;
+ this.withPaneInfo = transform.withPaneInfo;
+ }
+
+ @ProcessElement
+ public void processElement(
+ @Element T element,
+ @DoFn.Timestamp Instant timestamp,
+ BoundedWindow window,
+ PaneInfo paneInfo,
Review Comment:
false positive
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org