stop supervisor in case of metadata failures

[pjain1]

Possibly Fixes #3455

[b-slim]

not sure why not doing is gracefully by having true instead of false?

[pjain1]

If we stop it gracefully that means tasks that may have been started by this supervisor are allowed to publish what they have read which might be really tiny data as the metadata insert happens just after the supervisor starts.

Ideally supervisor.start() should be called only if metadata insert succeeds but it is not possible here.

[pjain1]

I checked the code again and stop gracefully set to false means running tasks are not killed. So, false flag is meant to handle cases like leadership changes. Here, we would like to just kill the tasks so doing what @cheddar suggested would be correct.

[cheddar]

I notice this comment here. I wonder if the supervisor shouldn't have a .verify() call on it to make sure the spec is legit, then persist to DB then start()?

Or, maybe just store in the DB first, start() and if start() doesn't work, fail out and remove from the DB again?

[pjain1]

updated the fix to start the supervisor first and then persist the spec if it succeeds otherwise write tombstone to prevent it being from picked up again.
@dclim does this make sense ?

[b-slim]

wondering is this needs to have a retry for some kind of transitory exception or just fail the task. I think with metadata storage we have in place some retry mechanism.

[pjain1]

I think it is ok for now .. i don't think we have retry mechanism at least for this insert operation

[b-slim]

fix seems legit, it will be nice if we can gracefully retry if the insert has a transient kind of issues. 👍

[himanshug]

does supervisor.start() failure guarantee that things are not left in a corrupted state and supervisor.stop(true) does not need to be called?

@dclim ?

[dclim]

@himanshug to be safe, we should probably make a call to consumer.close() in KafkaSupervisor.start() if the Kafka consumer was instantiated but an exception was thrown afterwards; @pjain1 do you mind adding this?

[dclim]

@pjain1 yes that seems good to me, sorry for missing your comment; logic looks good to me, maybe just ensure the Kafka consumer gets cleaned up if start() fails which is something I should've done before, thanks!

[pjain1]

@dclim added please verify

[dclim]

@pjain1 Can you actually move the code below it:

      firstRunTime = DateTime.now().plus(ioConfig.getStartDelay());
      scheduledExec.scheduleAtFixedRate(
          buildRunTask(),
          ioConfig.getStartDelay().getMillis(),
          Math.max(ioConfig.getPeriod().getMillis(), MAX_RUN_FREQUENCY_MILLIS),
          TimeUnit.MILLISECONDS
      );

      started = true;
      log.info(
          "Started KafkaSupervisor[%s], first run in [%s], with spec: [%s]",
          dataSource,
          ioConfig.getStartDelay(),
          spec.toString()
      );

into the try-catch as well? And also you should probably do a (if consumer != null) {consumer.close();} to avoid an NPE

[pjain1]

done

[dclim]

@pjain1 sweet thanks, LGTM 👍

[pjain1]

rebased with master

[pjain1]

@dclim @b-slim @himanshug can we get this merged for 0.9.2-rc2

[b-slim]

looks like we have 2 +1

[fjy]

@pjain1 backport plz

[pjain1]

@fjy #3532