Hi Team,
I have 8 nodes running Frontend, GMS, MAE and MCE.
I have scheduled ingestion to run at 10 AM daily. It is triggering the ingestion 8 times. Basically, the job is triggered from each GMS node.
Any option to fix this?
Attached ingestion run screenshot
Hey there! 
Make sure your message includes the following information if relevant, so we can help more effectively!
<@U06TM7M40RK> Can you help with the questions in this thread?
Please continue to tag RunLLM (e.g <@U06TM7M40RK>) to have it continue generating responses! We’re experimenting with a new AI assistant for questions in the <#CUMUWQU66> channel to help you track down helpful resources more quickly. Keep in mind that responses may not always be spot-on; feel free to ask it follow-up questions and don’t forget to provide its responses a 
or 
so other folks in the Community know if you’re still looking for help on the topic and to help us improve the assistant!
To address the issue of multiple ingestion triggers, it is important to understand the setup and configuration of your DataHub deployment. Here are a few questions to help diagnose the problem:
Based on the information provided, it seems that the ingestion job is being triggered from each GMS node, leading to multiple runs. This is a known issue when running multiple GMS instances or multiple MAE/MCE consumers.
Potential Solutions:
Standalone Mode Configuration:
Environment Variables:
MCP_CONSUMER_ENABLED, MCL_CONSUMER_ENABLED, and PE_CONSUMER_ENABLED variables are correctly set. Refer to the <Deployment Environment Variables | DataHub environment variables documentation> for more details.IngestionScheduler Singleton:
IngestionScheduler singleton class is responsible for scheduling ingestion runs. Ensure that only one instance of this class is running. This can be achieved by configuring the GMS deployment correctly. Refer to <Slack Slack discussion> for more details.Would you like more detailed steps on how to configure these settings?
Sources:
<@U06TM7M40RK>
Where is the IngestionScheduler singleton class located?
The IngestionScheduler singleton class is located in the DataHub codebase. Specifically, you can find it in the following file:
ingestion-scheduler/src/main/java/com/datahub/metadata/ingestion/IngestionScheduler.javaThe IngestionScheduler class is responsible for scheduling ingestion runs for ingestion sources defined within DataHub. It manages storing and triggering ingestion sources on a pre-defined schedule based on the information present in the DataHubIngestionSourceInfo aspect.
Here is a brief overview of the class:
package com.datahub.metadata.ingestion;
import com.linkedin.entity.client.EntityClient;
import io.datahubproject.metadata.context.OperationContext;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@Slf4j
@RequiredArgsConstructor
public class IngestionScheduler {
private final OperationContext systemOpContext;
private final EntityClient entityClient;
private final Map<Urn, ScheduledFuture<?>> nextIngestionSourceExecutionCache = new HashMap<>();
private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
public void init() {
// Initialization logic
}
public void unscheduleNextIngestionSourceExecution(final Urn ingestionSourceUrn) {
// Unschedule logic
}
public void unscheduleAll() {
// Unschedule all logic
}
public void scheduleNextIngestionSourceExecution(final Urn ingestionSourceUrn, final DataHubIngestionSourceInfo newInfo) {
// Schedule next execution logic
}
// Other methods and inner classes
}
Would you like more detailed information on how to configure or troubleshoot the IngestionScheduler?
Sources:
0 button 
0 button 
Hint: Mention @runllm in the thread for followups.<@U06TM7M40RK>
What triggers the scheduled ingestion?
The scheduled ingestion in DataHub is triggered by the IngestionScheduler class. This class is responsible for managing and executing scheduled ingestion runs based on the schedule defined in the DataHubIngestionSourceInfo aspect.
How Scheduled Ingestion is Triggered:
Initialization:
IngestionScheduler class initializes by loading all the ingestion source schedules and setting up the necessary tasks to trigger ingestion runs at the specified times.init method, which sets up a recurring task to refresh the schedules and schedule the next ingestion runs.Scheduling the Next Execution:
scheduleNextIngestionSourceExecution method computes the next execution time for an ingestion source based on its schedule and sets up a task to trigger the ingestion run at that time.ScheduledExecutorService to schedule the execution of the ingestion run.Execution Request:
ExecutionRequestRunnable is executed. This runnable creates an ExecutionRequest entity, which triggers the actual ingestion process.ExecutionRequestRunnable is responsible for creating a new DataHubExecutionRequest entity instance using an EntityClient.Here is a simplified code snippet to illustrate the process:
public class IngestionScheduler {
private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
private final Map<Urn, ScheduledFuture<?>> nextIngestionSourceExecutionCache = new HashMap<>();
public void init() {
// Schedule a recurring task to refresh schedules
scheduledExecutorService.scheduleAtFixedRate(
new BatchRefreshSchedulesRunnable(),
initialDelay,
refreshInterval,
TimeUnit.SECONDS);
}
public void scheduleNextIngestionSourceExecution(final Urn ingestionSourceUrn, final DataHubIngestionSourceInfo newInfo) {
// Compute the next execution time
ZonedDateTime nextExecDate = computeNextExecutionTime(newInfo.getSchedule());
long scheduleTime = nextExecDate.toInstant().toEpochMilli() - ZonedDateTime.now().toInstant().toEpochMilli();
// Schedule the ingestion run
ScheduledFuture<?> scheduledFuture = scheduledExecutorService.schedule(
new ExecutionRequestRunnable(ingestionSourceUrn, newInfo),
scheduleTime,
TimeUnit.MILLISECONDS);
nextIngestionSourceExecutionCache.put(ingestionSourceUrn, scheduledFuture);
}
private ZonedDateTime computeNextExecutionTime(DataHubIngestionSourceSchedule schedule) {
// Compute the next execution time based on the schedule
CronExpression generator = CronExpression.parse(schedule.getInterval());
return generator.next(ZonedDateTime.now());
}
}
Key Components:
ExecutionRequest to trigger the ingestion process.Would you like more detailed information on any specific part of this process?
Sources:
0 button 0 button Hint: Mention @runllm in the thread for followups.<@U06TM7M40RK>
Is it done by MAE or MCE?
The scheduled ingestion process itself is not directly managed by Metadata Audit Events (MAE) or Metadata Change Events (MCE). Instead, it is managed by the IngestionScheduler class, which schedules and triggers ingestion runs based on predefined schedules.
However, MAE and MCE play roles in the broader DataHub architecture, particularly in how metadata changes are processed and propagated. Here’s a breakdown of their roles:
Metadata Change Event (MCE):
Metadata Audit Event (MAE):
Scheduled Ingestion:
IngestionScheduler class.IngestionScheduler schedules and triggers ingestion runs based on the schedule defined in the DataHubIngestionSourceInfo aspect.ExecutionRequestRunnable is executed, which creates an ExecutionRequest entity to trigger the ingestion process.How They Interact:
Example Workflow:
Would you like more detailed information on any specific part of this process?
Sources:
0 button 0 button Hint: Mention @runllm in the thread for followups.<@U06TM7M40RK>
To prevent duplicate scheduled ingestions runs what consumer should have 1 replica?
To prevent duplicate scheduled ingestion runs, you should ensure that the MCE (Metadata Change Event) consumer has only one replica. This is because the MCE consumer is responsible for processing Metadata Change Proposals (MCPs) and triggering ingestion jobs. Having multiple replicas of the MCE consumer can lead to multiple ingestion triggers, as each replica may independently process the same events.
Steps to Configure MCE Consumer with a Single Replica:
Update Helm Chart:
replicaCount for the MCE consumer to 1.datahub-mce-consumer:
replicaCount: 1
Update Docker Compose:
replicas field for the MCE consumer service is set to 1.version: '3.7'
services:
datahub-mce-consumer:
image: datahub-mce-consumer:latest
container_name: datahub-mce-consumer
environment:
- ELASTICSEARCH_HOST=elasticsearch
- ELASTICSEARCH_PORT=9200
- ELASTICSEARCH_PROTOCOL=http
deploy:
replicas: 1
Kubernetes Deployment:
replicas field in the MCE consumer deployment YAML is set to 1.apiVersion: apps/v1
kind: Deployment
metadata:
name: datahub-mce-consumer
spec:
replicas: 1
template:
spec:
containers:
- name: datahub-mce-consumer
image: datahub-mce-consumer:latest
env:
- name: ELASTICSEARCH_HOST
value: "elasticsearch"
- name: ELASTICSEARCH_PORT
value: "9200"
- name: ELASTICSEARCH_PROTOCOL
value: "http"
Additional Considerations:
IngestionScheduler should be managed in a way that only one instance is responsible for scheduling ingestion runs.Would you like more detailed information on any specific part of this configuration?
Sources:
0 button 0 button Hint: Mention @runllm in the thread for followups.<@U06TM7M40RK>
How to configure GMS to avoid triggering multiple ingestion runs?
To configure the Global Metadata Service (GMS) to avoid triggering multiple ingestion runs, you need to ensure that the ingestion scheduling is managed in a way that only one instance of the IngestionScheduler is responsible for scheduling ingestion runs. Here are the steps to achieve this:
Steps to Configure GMS:
Standalone Mode for MCE Consumer:
Disable Ingestion Scheduling on Additional GMS Instances:
IngestionScheduler on additional instances.Example Configuration:
Helm Chart Configuration:
datahub-gms:
replicaCount: 3
extraEnv:
- name: INGESTION_SCHEDULER_ENABLED
value: "false"
INGESTION_SCHEDULER_ENABLED environment variable to true only on one GMS instance.Kubernetes Deployment Configuration:
INGESTION_SCHEDULER_ENABLED environment variable.apiVersion: apps/v1
kind: Deployment
metadata:
name: datahub-gms
spec:
replicas: 3
template:
spec:
containers:
- name: datahub-gms
image: datahub-gms:latest
env:
- name: INGESTION_SCHEDULER_ENABLED
value: "false"
INGESTION_SCHEDULER_ENABLED environment variable to true.apiVersion: apps/v1
kind: Deployment
metadata:
name: datahub-gms-scheduler
spec:
replicas: 1
template:
spec:
containers:
- name: datahub-gms
image: datahub-gms:latest
env:
- name: INGESTION_SCHEDULER_ENABLED
value: "true"
Docker Compose Configuration:
INGESTION_SCHEDULER_ENABLED environment variable is set to false for additional GMS instances.version: '3.7'
services:
datahub-gms:
image: datahub-gms:latest
environment:
- INGESTION_SCHEDULER_ENABLED=false
deploy:
replicas: 3
datahub-gms-scheduler:
image: datahub-gms:latest
environment:
- INGESTION_SCHEDULER_ENABLED=true
deploy:
replicas: 1
Additional Considerations:
By following these steps, you can configure GMS to avoid triggering multiple ingestion runs and ensure that only one instance is responsible for scheduling ingestion runs.
Would you like more detailed steps on any specific part of this configuration?
Sources:
0 button 0 button Hint: Mention @runllm in the thread for followups.<@U06TM7M40RK>
INGESTION_SCHEDULER_ENABLED is not present in the codebase
Thank you for pointing that out. If the INGESTION_SCHEDULER_ENABLED environment variable is not present in the codebase, we need to take a different approach to ensure that only one instance of the IngestionScheduler is responsible for scheduling ingestion runs.
Alternative Approach:
Leader Election:
Custom Configuration:
IngestionScheduler class to check for a custom configuration or environment variable that determines whether it should run the scheduler.Example Implementation:
Leader Election with Kubernetes:
Install the Leader Election Library:
kubernetes-client to implement leader election in your GMS deployment.Modify the IngestionScheduler Class:
IngestionScheduler class to participate in leader election and only run the scheduler if it is the leader.import io.kubernetes.client.extended.leaderelection.LeaderElectionConfig;
import io.kubernetes.client.extended.leaderelection.LeaderElector;
import io.kubernetes.client.extended.leaderelection.resourcelock.LeaseLock;
import io.kubernetes.client.util.credentials.AccessTokenAuthentication;
import io.kubernetes.client.util.credentials.ClientCertificateAuthentication;
import io.kubernetes.client.util.credentials.KubeconfigAuthentication;
import io.kubernetes.client.util.credentials.TokenFileAuthentication;
import java.time.Duration;
public class IngestionScheduler {
private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
private final Map<Urn, ScheduledFuture<?>> nextIngestionSourceExecutionCache = new HashMap<>();
public void init() {
// Leader election configuration
LeaseLock leaseLock = new LeaseLock("default", "datahub-gms-leader", "datahub-gms");
LeaderElectionConfig config = new LeaderElectionConfig(
leaseLock,
Duration.ofMillis(10000), // Lease duration
Duration.ofMillis(8000), // Renew deadline
Duration.ofMillis(2000) // Retry period
);
LeaderElector leaderElector = new LeaderElector(config);
leaderElector.run(
() -> {
// This block will only be executed by the leader
scheduledExecutorService.scheduleAtFixedRate(
new BatchRefreshSchedulesRunnable(),
initialDelay,
refreshInterval,
TimeUnit.SECONDS);
}
);
}
public void scheduleNextIngestionSourceExecution(final Urn ingestionSourceUrn, final DataHubIngestionSourceInfo newInfo) {
// Compute the next execution time
ZonedDateTime nextExecDate = computeNextExecutionTime(newInfo.getSchedule());
long scheduleTime = nextExecDate.toInstant().toEpochMilli() - ZonedDateTime.now().toInstant().toEpochMilli();
// Schedule the ingestion run
ScheduledFuture<?> scheduledFuture = scheduledExecutorService.schedule(
new ExecutionRequestRunnable(ingestionSourceUrn, newInfo),
scheduleTime,
TimeUnit.MILLISECONDS);
nextIngestionSourceExecutionCache.put(ingestionSourceUrn, scheduledFuture);
}
private ZonedDateTime computeNextExecutionTime(DataHubIngestionSourceSchedule schedule) {
// Compute the next execution time based on the schedule
CronExpression generator = CronExpression.parse(schedule.getInterval());
return generator.next(ZonedDateTime.now());
}
}
IngestionScheduler.Custom Configuration:
IngestionScheduler class to determine whether it should run the scheduler.public class IngestionScheduler {
private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
private final Map<Urn, ScheduledFuture<?>> nextIngestionSourceExecutionCache = new HashMap<>();
private final boolean isSchedulerEnabled;
public IngestionScheduler(boolean isSchedulerEnabled) {
this.isSchedulerEnabled = isSchedulerEnabled;
}
public void init() {
if (isSchedulerEnabled) {
// Schedule a recurring task to refresh schedules
scheduledExecutorService.scheduleAtFixedRate(
new BatchRefreshSchedulesRunnable(),
initialDelay,
refreshInterval,
TimeUnit.SECONDS);
}
}
public void scheduleNextIngestionSourceExecution(final Urn ingestionSourceUrn, final DataHubIngestionSourceInfo newInfo) {
if (isSchedulerEnabled) {
// Compute the next execution time
ZonedDateTime nextExecDate = computeNextExecutionTime(newInfo.getSchedule());
long scheduleTime = nextExecDate.toInstant().toEpochMilli() - ZonedDateTime.now().toInstant().toEpochMilli();
// Schedule the ingestion run
ScheduledFuture<?> scheduledFuture = scheduledExecutorService.schedule(
new ExecutionRequestRunnable(ingestionSourceUrn, newInfo),
scheduleTime,
TimeUnit.MILLISECONDS);
nextIngestionSourceExecutionCache.put(ingestionSourceUrn, scheduledFuture);
}
}
private ZonedDateTime computeNextExecutionTime(DataHubIngestionSourceSchedule schedule) {
// Compute the next execution time based on the schedule
CronExpression generator = CronExpression.parse(schedule.getInterval());
return generator.next(ZonedDateTime.now());
}
}
isSchedulerEnabled flag to true for only one GMS instance and false for the others.Would you like more detailed steps on implementing any of these solutions?
Sources:
0 button 0 button Hint: Mention @runllm in the thread for followups.