Skip to content

[GOBBLIN-1830] Improving Container Transition Tracking in Streaming Data Ingestion #3693

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ZihanLi58 merged 5 commits into from
May 19, 2023
Merged

[GOBBLIN-1830] Improving Container Transition Tracking in Streaming Data Ingestion #3693

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
9af584c
address comments
Mar 25, 2022
a74ca18
use connectionmanager when httpclient is not cloesable
Mar 25, 2022
675f728
[GOBBLIN-1830]Improving Container Transition Tracking in Streaming Da…
May 3, 2023
be8a559
emmit event with a different name
May 9, 2023
11c93f3
remove unnecessary log
May 9, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions ...in/java/org/apache/gobblin/source/extractor/extract/kafka/KafkaExtractorStatsTracker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
import org.apache.gobblin.configuration.WorkUnitState;
import org.apache.gobblin.metrics.MetricContext;
import org.apache.gobblin.metrics.event.EventSubmitter;
import org.apache.gobblin.metrics.event.GobblinEventBuilder;
import org.apache.gobblin.runtime.api.TaskEventMetadataGenerator;
import org.apache.gobblin.util.TaskEventMetadataUtils;

Expand All @@ -56,6 +57,7 @@ public class KafkaExtractorStatsTracker {

private static final String EMPTY_STRING = "";
private static final String GOBBLIN_KAFKA_NAMESPACE = "gobblin.kafka";
private static final String KAFKA_EXTRACTOR_CONTAINER_TRANSITION_EVENT_NAME = "KafkaExtractorContainerTransitionEvent";
private static final String KAFKA_EXTRACTOR_TOPIC_METADATA_EVENT_NAME = "KafkaExtractorTopicMetadata";
private static final String LOW_WATERMARK = "lowWatermark";
private static final String ACTUAL_HIGH_WATERMARK = "actualHighWatermark";
Expand Down Expand Up @@ -497,6 +499,21 @@ public void emitTrackingEvents(MetricContext context, Map<KafkaPartition, Map<St
}
}

/**
* Emit Tracking events reporting the topic partition information this extractor handled to be consumed by a monitoring application.
* @param context the current {@link MetricContext}
*/
public void submitEventToIndicateContainerTransition(MetricContext context) {
for (int i = 0; i < this.partitions.size(); i++) {
KafkaPartition partitionKey = this.partitions.get(i);
GobblinEventBuilder gobblinEventBuilder = new GobblinEventBuilder(KAFKA_EXTRACTOR_CONTAINER_TRANSITION_EVENT_NAME, GOBBLIN_KAFKA_NAMESPACE);
gobblinEventBuilder.addMetadata(TOPIC, partitionKey.getTopicName());
gobblinEventBuilder.addMetadata(PARTITION, Integer.toString(partitionKey.getId()));
gobblinEventBuilder.addAdditionalMetadata(this.taskEventMetadataGenerator.getMetadata(workUnitState, KAFKA_EXTRACTOR_CONTAINER_TRANSITION_EVENT_NAME));
EventSubmitter.submit(context, gobblinEventBuilder);
}
}

/**
* A helper function to merge tags for KafkaPartition. Separate into a package-private method for ease of testing.
*/
Expand Down
8 changes: 8 additions & 0 deletions .../main/java/org/apache/gobblin/source/extractor/extract/kafka/KafkaStreamingExtractor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -246,6 +246,8 @@ public KafkaStreamingExtractor(WorkUnitState state) {
state.getPropAsLong(KAFKA_EXTRACTOR_STATS_REPORTING_INTERVAL_MINUTES_KEY,
DEFAULT_KAFKA_EXTRACTOR_STATS_REPORTING_INTERVAL_MINUTES) * 60 * 1000;
resetExtractorStatsAndWatermarks(true);
//Even though we haven't start ingesting yet, emit event to indicate the container transition.
submitEventToIndicateContainerTransition();

//Schedule a thread for reporting Kafka consumer metrics
this.scheduledExecutorService.scheduleAtFixedRate(() -> {
Expand All @@ -271,6 +273,12 @@ public KafkaStreamingExtractor(WorkUnitState state) {
this.workUnitState.getProp(KafkaSource.RECORD_CREATION_TIMESTAMP_UNIT, TimeUnit.MILLISECONDS.name()));
}

private void submitEventToIndicateContainerTransition() {
if (this.isInstrumentationEnabled()) {
this.statsTracker.submitEventToIndicateContainerTransition(getMetricContext());
}
}

private Map<KafkaPartition, LongWatermark> getTopicPartitionWatermarks(List<KafkaPartition> topicPartitions) {
List<String> topicPartitionStrings =
topicPartitions.stream().map(topicPartition -> topicPartition.toString()).collect(Collectors.toList());
Expand Down