You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by GitBox <gi...@apache.org> on 2021/04/28 00:05:45 UTC
[GitHub] [gobblin] arjun4084346 commented on a change in pull request #3268: [GOBBLIN-1437] Segregate FlowConfigs/Delete and FlowExecutions/Delete functions
arjun4084346 commented on a change in pull request #3268:
URL: https://github.com/apache/gobblin/pull/3268#discussion_r621681486
##########
File path: gobblin-runtime/src/main/java/org/apache/gobblin/runtime/job_monitor/KafkaJobMonitor.java
##########
@@ -100,17 +100,29 @@ protected void shutdownMetrics()
@Override
protected void processMessage(DecodeableKafkaRecord<byte[],byte[]> message) {
try {
- Collection<Either<JobSpec, URI>> parsedCollection = parseJobSpec(message.getValue());
- for (Either<JobSpec, URI> parsedMessage : parsedCollection) {
- if (parsedMessage instanceof Either.Left) {
- this.newSpecs.inc();
- this.jobCatalog.put(((Either.Left<JobSpec, URI>) parsedMessage).getLeft());
- } else if (parsedMessage instanceof Either.Right) {
- this.removedSpecs.inc();
- URI jobSpecUri = ((Either.Right<JobSpec, URI>) parsedMessage).getRight();
- this.jobCatalog.remove(jobSpecUri, true);
- // Delete the job state if it is a delete spec request
- deleteStateStore(jobSpecUri);
+ Collection<JobSpec> parsedCollection = parseJobSpec(message.getValue());
+ for (JobSpec parsedMessage : parsedCollection) {
+ SpecExecutor.Verb verb = SpecExecutor.Verb.valueOf(parsedMessage.getMetadata().get(SpecExecutor.VERB_KEY));
+
+ switch (verb) {
+ case ADD:
+ case UPDATE:
+ case UNKNOWN: // unknown are considered as add request to maintain backward compatibility
+ this.newSpecs.inc();
+ this.jobCatalog.put(parsedMessage);
+ break;
+ case DELETE:
+ this.removedSpecs.inc();
+ URI jobSpecUri = parsedMessage.getUri();
+ this.jobCatalog.remove(jobSpecUri, true);
+ // Delete the job state if it is a delete spec request
+ deleteStateStore(jobSpecUri);
Review comment:
kafka offsets are committed in HighLevelConsumer only after this method returns. So if state store deletion is not completed, both 'removal of job' and 'state store deletion' will be retried whenever the service restarts.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org