KafkaIndexTask can delete published segments on restart

[jihoonson]

This can happen in the following scenario.

1. A kafka index task starts publishing segments.
2. The task succeeds to publish segments and is stopped immediately (by restarting the machine).
3. When the task is restored, it restores all sequences it kept in memory before restarting.
4. After reading some more events from Kafka, the task tries to publish segments. These segments include the ones which were published before restarting because the restored sequences contain them.
5. Since the segments which are published twice are already stored in metastore, the publish fails.
6. The set of published segments in metastore is different from the set of segments the task is trying because the task read more data.
7. The task thinks that the publish actually failed and removes the published segments from deep storage.

[gianm]

5. Since the segments which are published twice are already stored in metastore, the publish fails.

This doesn't seem right: there is code specifically to handle the case where a task tries to publish a segment that some task already published. It happens all the time with replicas, and they just ignore that segment and move on to the next one.

I wonder if the real reason for publish failure is that the startMetadata doesn't match up. I bet it wouldn't match up: it sounds like the task is trying to publish from the point it originally started from rather than from the point it last published.

It sounds like there are two separate problems here:

• In (7) the task should not have removed the published segments (this is the biggest bug).
• In (3) the task should have done something smarter instead of restoring a setup that couldn't possibly work out.

[jihoonson]

This doesn't seem right: there is code specifically to handle the case where a task tries to publish a segment that some task already published. It happens all the time with replicas, and they just ignore that segment and move on to the next one.

Yes, correct. The task doesn't fail at this point, but it just fails to update the metastore.

I wonder if the real reason for publish failure is that the startMetadata doesn't match up. I bet it wouldn't match up: it sounds like the task is trying to publish from the point it originally started from rather than from the point it last published.

Here is the code what this error happens.

              final boolean published = publisher.publishSegments(
                  ImmutableSet.copyOf(segmentsAndMetadata.getSegments()),
                  metadata == null ? null : ((AppenderatorDriverMetadata) metadata).getCallerMetadata()
              );

              if (published) {
                log.info("Published segments.");
              } else {
                log.info("Transaction failure while publishing segments, checking if someone else beat us to it.");
                final Set<SegmentIdentifier> segmentsIdentifiers = segmentsAndMetadata
                    .getSegments()
                    .stream()
                    .map(SegmentIdentifier::fromDataSegment)
                    .collect(Collectors.toSet());
                if (usedSegmentChecker.findUsedSegments(segmentsIdentifiers)
                                      .equals(Sets.newHashSet(segmentsAndMetadata.getSegments()))) {
                  log.info(
                      "Removing our segments from deep storage because someone else already published them: %s",
                      segmentsAndMetadata.getSegments()
                  );
                  segmentsAndMetadata.getSegments().forEach(dataSegmentKiller::killQuietly);

                  log.info("Our segments really do exist, awaiting handoff.");
                } else {
                  throw new ISE("Failed to publish segments[%s]", segmentsAndMetadata.getSegments());
                }
              }

published is false, so the task checks the segments it failed to publish are already being used using usedSegmentChecker which is the ActionBasedUsedSegmentChecker in this case. And usedSegmentChecker.findUsedSegments(segmentsIdentifiers).equals(Sets.newHashSet(segmentsAndMetadata.getSegments())) returns false and the task throws an exception. This is because the task generated more segments after restarting. I've verified this by comparing the publishing segments before and after restart.

I didn't see any logs related to metadata mismatch.

In (7) the task should not have removed the published segments (this is the biggest bug).

Not sure about this. This can make publishing segments non-atomic as well as potential garbage segments. Probably solving (3) would be enough because this should never happen?

In (3) the task should have done something smarter instead of restoring a setup that couldn't possibly work out.

Yes, I think we should keep the task states in local disk which represents what the task was doing. Also, for usedSegmentChecker.findUsedSegments(segmentsIdentifiers).equals(Sets.newHashSet(segmentsAndMetadata.getSegments())), we can change this to check usedSegments include segmentsAndMetadata.getSegments() and continue publishing the segments not in usedSegments.

[gianm]

I think this may be fixed by #6155, but would like to double-check that there are no lingering issues due to the fact that the sequences persist file and the appenderator persist file are separate files.

[jihoonson]

To make it clear, #6155 fixes Kafka tasks to not delete published segments any more. In the same scenario, the restored Kafka tasks would fail and the supervisor would respawn the new tasks which will finish the job.

[gianm]

Got it, I think we can close this then.

[jihoonson]

@gianm I think it makes sense to close this issue since the title is about kafka tasks deleting pushed segments. However, the tasks would still fail in the same scenario which can make users confused. Do you think we need to file this in another issue?