You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by "Blazer-007 (via GitHub)" <gi...@apache.org> on 2025/11/19 15:56:10 UTC
[PR] [GOBBLIN-2239] introduce pushMessagesSync for sync GTE emission [gobblin]
Blazer-007 opened a new pull request, #4156:
URL: https://github.com/apache/gobblin/pull/4156
Dear Gobblin maintainers,
Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!
### JIRA
- [ ] My PR addresses the following [Gobblin JIRA](https://issues.apache.org/jira/browse/GOBBLIN/) issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-2239
### Description
- [ ] Here are some details about my PR, including screenshots (if applicable):
In Gobblin-on-Temporal, GTE can be emitted from any container in case of more than 1 containers if event queue has very less GTE those are flushed while closing only and sometimes due to early SIGKILL sent from Yarn NM GTE couldn't be flushed properly.
Have introduced `pushMessagesSync` in `Pusher.java` interface with default implementation to call async only for backward compatibility
### Tests
- [ ] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
NA
### Commits
- [ ] My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"
--
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: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2239] Introduce pushMessagesSync for sync GTE emission [gobblin]
Posted by "Blazer-007 (via GitHub)" <gi...@apache.org>.
Blazer-007 merged PR #4156:
URL: https://github.com/apache/gobblin/pull/4156
--
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: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2239] Introduce pushMessagesSync for sync GTE emission [gobblin]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4156:
URL: https://github.com/apache/gobblin/pull/4156#discussion_r2544295654
##########
gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/Pusher.java:
##########
@@ -27,7 +27,16 @@
public interface Pusher<M> extends Closeable {
/**
* Push all byte array messages to the Kafka topic.
- * @param messages List of byte array messages to push to Kakfa.
+ * @param messages List of byte array messages to push to Kafka.
*/
void pushMessages(List<M> messages);
+
+ /**
+ * Synchronous version of {@link #pushMessages(List)}.
+ * Default implementation just calls {@link #pushMessages(List)}.
+ * @param messages List of byte array messages to push to Kafka.
+ */
+ default void pushMessagesSync(List<M> messages) {
+ pushMessages(messages);
+ }
Review Comment:
The default implementation of `pushMessagesSync` only calls `pushMessages`, which doesn't actually provide synchronous behavior. None of the concrete `Pusher` implementations (KafkaKeyValueProducerPusher for Kafka 08, 09, and 1) override this method to call `producer.flush()`. This means the synchronous flushing functionality will not work as intended. Each KafkaKeyValueProducerPusher implementation should override `pushMessagesSync` to call `this.producer.flush()` after sending messages.
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/workflows/metrics/SubmitGTEActivityImpl.java:
##########
@@ -19,22 +19,50 @@
import java.util.Map;
-import io.temporal.workflow.Workflow;
-
-import org.slf4j.Logger;
+import java.util.Optional;
+import lombok.extern.slf4j.Slf4j;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import org.apache.gobblin.cluster.ContainerMetrics;
import org.apache.gobblin.metrics.event.EventSubmitter;
import org.apache.gobblin.metrics.event.GobblinEventBuilder;
import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.metrics.kafka.KafkaAvroEventKeyValueReporter;
+import org.apache.gobblin.temporal.GobblinTemporalConfigurationKeys;
+import org.apache.gobblin.temporal.cluster.WorkerConfig;
+import org.apache.gobblin.util.ConfigUtils;
+@Slf4j
public class SubmitGTEActivityImpl implements SubmitGTEActivity {
- private static Logger log = Workflow.getLogger(SubmitGTEActivityImpl.class);
@Override
public void submitGTE(GobblinEventBuilder eventBuilder, EventSubmitterContext eventSubmitterContext) {
- log.info("submitting GTE - {}", summarizeEventMetadataForLogging(eventBuilder));
+ log.info("Submitting GTE - {}", summarizeEventMetadataForLogging(eventBuilder));
eventSubmitterContext.create().submit(eventBuilder);
+ Config config = WorkerConfig.of(this).orElse(ConfigFactory.load());
Review Comment:
The config lookup happens on every GTE submission. Consider caching the config and ContainerMetrics instance as instance variables (initialized once, possibly lazily) to avoid repeated lookups, especially since SubmitGTEActivityImpl is registered as a singleton activity implementation and the config should remain constant throughout the activity's lifecycle.
##########
gobblin-metrics-libs/gobblin-metrics-base/src/main/java/org/apache/gobblin/metrics/reporter/EventReporter.java:
##########
@@ -163,12 +163,28 @@ public void report() {
reportEventQueue(this.reportingQueue);
}
+ /**
+ * Report all {@link org.apache.gobblin.metrics.GobblinTrackingEvent}s in the queue synchronously.
+ */
+ public void reportSynchronously() {
+ reportEventQueueSynchronously(this.reportingQueue);
+ }
+
/**
* Emit all {@link org.apache.gobblin.metrics.GobblinTrackingEvent} in queue.
* @param queue {@link java.util.Queue} containing {@link org.apache.gobblin.metrics.GobblinTrackingEvent}s that should be emitted.
*/
public abstract void reportEventQueue(Queue<GobblinTrackingEvent> queue);
+ /**
+ * Emit all {@link org.apache.gobblin.metrics.GobblinTrackingEvent} in queue synchronously.
+ * Default implementation just calls {@link #reportEventQueue(Queue)} for backward compatibility.
+ * @param queue {@link java.util.Queue} containing {@link org.apache.gobblin.metrics.GobblinTrackingEvent}s that should be emitted.
+ */
+ public void reportEventQueueSynchronously(Queue<GobblinTrackingEvent> queue) {
+ reportEventQueue(queue);
+ }
Review Comment:
The default implementation of `reportEventQueueSynchronously` should clarify in the Javadoc that it provides no actual synchronous guarantees and subclasses must override it to implement true synchronous behavior. Update the comment on line 181 to state: 'Default implementation provides no synchronous guarantees and just calls {@link #reportEventQueue(Queue)}. Subclasses should override this method to implement true synchronous reporting behavior.'
--
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: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2239] Introduce pushMessagesSync for sync GTE emission [gobblin]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4156:
URL: https://github.com/apache/gobblin/pull/4156#discussion_r2542767484
##########
gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/Pusher.java:
##########
@@ -30,4 +30,13 @@ public interface Pusher<M> extends Closeable {
* @param messages List of byte array messages to push to Kakfa.
*/
void pushMessages(List<M> messages);
+
+ /**
+ * Synchronous version of {@link #pushMessages(List)}.
+ * Default implementation just calls {@link #pushMessages(List)}.
+ * @param messages
Review Comment:
Missing JavaDoc parameter description. The `@param messages` tag should include a description of what messages are being pushed, similar to line 30.
```suggestion
* @param messages List of byte array messages to push to Kafka.
```
##########
gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/KafkaEventKeyValueReporter.java:
##########
@@ -54,6 +54,27 @@ protected KafkaEventKeyValueReporter(Builder<?> builder) throws IOException {
@Override
public void reportEventQueue(Queue<GobblinTrackingEvent> queue) {
+ List<Pair<String, byte[]>> events = getEventsFromQueue(queue);
+ if (!events.isEmpty()) {
+ log.info("Pushing {} Gobblin Tracking Events to Kafka", events.size());
+ this.kafkaPusher.pushMessages(events);
+ } else {
+ log.debug("No GTE to push.");
+ }
+ }
+
+ @Override
+ public void reportEventQueueSynchronously(Queue<GobblinTrackingEvent> queue) {
+ List<Pair<String, byte[]>> events = getEventsFromQueue(queue);
+ if (!events.isEmpty()) {
+ log.info("Pushing {} Gobblin Tracking Events to Kafka", events.size());
+ this.kafkaPusher.pushMessagesSync(events);
+ } else {
+ log.debug("No GTE to push.");
+ }
+ }
Review Comment:
Code duplication: The logic in `reportEventQueue()` and `reportEventQueueSynchronously()` is identical except for the call to `pushMessages()` vs `pushMessagesSync()`. Consider refactoring by having both methods call a common helper method that accepts the push method as a parameter, or have one method call the other with a boolean flag to indicate sync/async mode.
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/workflows/metrics/SubmitGTEActivityImpl.java:
##########
@@ -19,22 +19,50 @@
import java.util.Map;
-import io.temporal.workflow.Workflow;
-
-import org.slf4j.Logger;
+import java.util.Optional;
+import lombok.extern.slf4j.Slf4j;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import org.apache.gobblin.cluster.ContainerMetrics;
import org.apache.gobblin.metrics.event.EventSubmitter;
import org.apache.gobblin.metrics.event.GobblinEventBuilder;
import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.metrics.kafka.KafkaAvroEventKeyValueReporter;
+import org.apache.gobblin.temporal.GobblinTemporalConfigurationKeys;
+import org.apache.gobblin.temporal.cluster.WorkerConfig;
+import org.apache.gobblin.util.ConfigUtils;
+@Slf4j
public class SubmitGTEActivityImpl implements SubmitGTEActivity {
- private static Logger log = Workflow.getLogger(SubmitGTEActivityImpl.class);
@Override
public void submitGTE(GobblinEventBuilder eventBuilder, EventSubmitterContext eventSubmitterContext) {
- log.info("submitting GTE - {}", summarizeEventMetadataForLogging(eventBuilder));
+ log.info("Submitting GTE - {}", summarizeEventMetadataForLogging(eventBuilder));
eventSubmitterContext.create().submit(eventBuilder);
+ Config config = WorkerConfig.of(this).orElse(ConfigFactory.load());
+ if (config.hasPath(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_TASK_RUNNER_ID) &&
+ config.hasPath(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_APPLICATION_NAME)) {
+ String containerMetricsApplicationName = config.getString(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_APPLICATION_NAME);
+ String containerMetricsTaskRunnerId = config.getString(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_TASK_RUNNER_ID);
+ ContainerMetrics containerMetrics = ContainerMetrics.get(ConfigUtils.configToState(config), containerMetricsApplicationName, containerMetricsTaskRunnerId);
+ log.info("Fetched container metrics instance {} for taskRunnerId: {} and applicationName: {}", containerMetrics.toString(),
Review Comment:
Redundant use of `toString()`. When using string formatting with `{}` placeholders in SLF4J, the object's `toString()` method is called automatically. The explicit `.toString()` call is unnecessary and should be removed.
```suggestion
log.info("Fetched container metrics instance {} for taskRunnerId: {} and applicationName: {}", containerMetrics,
```
##########
gobblin-metrics-libs/gobblin-metrics-base/src/main/java/org/apache/gobblin/metrics/reporter/EventReporter.java:
##########
@@ -163,12 +163,28 @@ public void report() {
reportEventQueue(this.reportingQueue);
}
+ /**
+ * Report all {@link org.apache.gobblin.metrics.GobblinTrackingEvent}s in the queue synchronously.
+ */
+ public void reportSynchronously() {
+ reportEventQueueSynchronously(this.reportingQueue);
+ }
+
/**
* Emit all {@link org.apache.gobblin.metrics.GobblinTrackingEvent} in queue.
* @param queue {@link java.util.Queue} containing {@link org.apache.gobblin.metrics.GobblinTrackingEvent}s that should be emitted.
*/
public abstract void reportEventQueue(Queue<GobblinTrackingEvent> queue);
+ /**
+ * Emit all {@link org.apache.gobblin.metrics.GobblinTrackingEvent} in queue synchronously.
+ * Default implementation just calls {@link #reportEventQueue(Queue)} for backward compatibility.
+ * @param queue
Review Comment:
Missing JavaDoc parameter description. The `@param queue` tag should include a description of the queue parameter, similar to line 175.
```suggestion
* @param queue {@link java.util.Queue} containing {@link org.apache.gobblin.metrics.GobblinTrackingEvent}s that should be emitted synchronously.
```
##########
gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/Pusher.java:
##########
@@ -30,4 +30,13 @@ public interface Pusher<M> extends Closeable {
* @param messages List of byte array messages to push to Kakfa.
Review Comment:
Typo in JavaDoc: "Kakfa" should be "Kafka".
```suggestion
* @param messages List of byte array messages to push to Kafka.
```
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/workflows/metrics/SubmitGTEActivityImpl.java:
##########
@@ -19,22 +19,50 @@
import java.util.Map;
-import io.temporal.workflow.Workflow;
-
-import org.slf4j.Logger;
+import java.util.Optional;
+import lombok.extern.slf4j.Slf4j;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import org.apache.gobblin.cluster.ContainerMetrics;
import org.apache.gobblin.metrics.event.EventSubmitter;
import org.apache.gobblin.metrics.event.GobblinEventBuilder;
import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.metrics.kafka.KafkaAvroEventKeyValueReporter;
+import org.apache.gobblin.temporal.GobblinTemporalConfigurationKeys;
+import org.apache.gobblin.temporal.cluster.WorkerConfig;
+import org.apache.gobblin.util.ConfigUtils;
+@Slf4j
public class SubmitGTEActivityImpl implements SubmitGTEActivity {
- private static Logger log = Workflow.getLogger(SubmitGTEActivityImpl.class);
@Override
public void submitGTE(GobblinEventBuilder eventBuilder, EventSubmitterContext eventSubmitterContext) {
- log.info("submitting GTE - {}", summarizeEventMetadataForLogging(eventBuilder));
+ log.info("Submitting GTE - {}", summarizeEventMetadataForLogging(eventBuilder));
eventSubmitterContext.create().submit(eventBuilder);
+ Config config = WorkerConfig.of(this).orElse(ConfigFactory.load());
+ if (config.hasPath(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_TASK_RUNNER_ID) &&
+ config.hasPath(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_APPLICATION_NAME)) {
+ String containerMetricsApplicationName = config.getString(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_APPLICATION_NAME);
+ String containerMetricsTaskRunnerId = config.getString(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_TASK_RUNNER_ID);
+ ContainerMetrics containerMetrics = ContainerMetrics.get(ConfigUtils.configToState(config), containerMetricsApplicationName, containerMetricsTaskRunnerId);
+ log.info("Fetched container metrics instance {} for taskRunnerId: {} and applicationName: {}", containerMetrics.toString(),
+ containerMetricsTaskRunnerId, containerMetricsApplicationName);
+ Optional<KafkaAvroEventKeyValueReporter> kafkaReporterOptional = containerMetrics.getScheduledReporter(KafkaAvroEventKeyValueReporter.class);
+ if (kafkaReporterOptional.isPresent()) {
+ log.info("Submitting GTE in synchronous manner to Kafka reporter");
+ kafkaReporterOptional.get().reportSynchronously();
+ log.info("Submitted GTE to Kafka reporter");
+ } else {
+ log.error("No KafkaAvroEventKeyValueReporter found in container metrics for taskRunnerId: {} and applicationName: {}",
Review Comment:
Logging at ERROR level may not be appropriate here. If the KafkaAvroEventKeyValueReporter is optional, this should be logged at WARN or INFO level. ERROR level should be reserved for actual failure conditions that prevent the operation from completing successfully.
```suggestion
log.warn("No KafkaAvroEventKeyValueReporter found in container metrics for taskRunnerId: {} and applicationName: {}",
```
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/workflows/metrics/SubmitGTEActivityImpl.java:
##########
@@ -19,22 +19,50 @@
import java.util.Map;
-import io.temporal.workflow.Workflow;
-
-import org.slf4j.Logger;
+import java.util.Optional;
+import lombok.extern.slf4j.Slf4j;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import org.apache.gobblin.cluster.ContainerMetrics;
import org.apache.gobblin.metrics.event.EventSubmitter;
import org.apache.gobblin.metrics.event.GobblinEventBuilder;
import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.metrics.kafka.KafkaAvroEventKeyValueReporter;
+import org.apache.gobblin.temporal.GobblinTemporalConfigurationKeys;
+import org.apache.gobblin.temporal.cluster.WorkerConfig;
+import org.apache.gobblin.util.ConfigUtils;
+@Slf4j
public class SubmitGTEActivityImpl implements SubmitGTEActivity {
- private static Logger log = Workflow.getLogger(SubmitGTEActivityImpl.class);
@Override
public void submitGTE(GobblinEventBuilder eventBuilder, EventSubmitterContext eventSubmitterContext) {
- log.info("submitting GTE - {}", summarizeEventMetadataForLogging(eventBuilder));
+ log.info("Submitting GTE - {}", summarizeEventMetadataForLogging(eventBuilder));
eventSubmitterContext.create().submit(eventBuilder);
+ Config config = WorkerConfig.of(this).orElse(ConfigFactory.load());
+ if (config.hasPath(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_TASK_RUNNER_ID) &&
+ config.hasPath(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_APPLICATION_NAME)) {
+ String containerMetricsApplicationName = config.getString(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_APPLICATION_NAME);
+ String containerMetricsTaskRunnerId = config.getString(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_TASK_RUNNER_ID);
+ ContainerMetrics containerMetrics = ContainerMetrics.get(ConfigUtils.configToState(config), containerMetricsApplicationName, containerMetricsTaskRunnerId);
+ log.info("Fetched container metrics instance {} for taskRunnerId: {} and applicationName: {}", containerMetrics.toString(),
+ containerMetricsTaskRunnerId, containerMetricsApplicationName);
+ Optional<KafkaAvroEventKeyValueReporter> kafkaReporterOptional = containerMetrics.getScheduledReporter(KafkaAvroEventKeyValueReporter.class);
+ if (kafkaReporterOptional.isPresent()) {
+ log.info("Submitting GTE in synchronous manner to Kafka reporter");
+ kafkaReporterOptional.get().reportSynchronously();
+ log.info("Submitted GTE to Kafka reporter");
+ } else {
+ log.error("No KafkaAvroEventKeyValueReporter found in container metrics for taskRunnerId: {} and applicationName: {}",
+ containerMetricsTaskRunnerId, containerMetricsApplicationName);
+ }
+ } else {
+ log.error("Both {} and {} must be set to fetch container metrics instance for synchronous GTE submission",
Review Comment:
Logging at ERROR level may not be appropriate here. If these configuration keys are optional (as suggested by the conditional check), then their absence should be logged at WARN or INFO level instead. ERROR level typically indicates a problem that prevents normal operation, but the code continues without these values.
```suggestion
log.warn("Both {} and {} should be set to fetch container metrics instance for synchronous GTE submission",
```
--
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: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org