Fix Kafka Indexing task pause forever if no events in taskDuration (#5656)

[surekhasaharan]

If the endOffset is same as startOffset, still let the task resume instead of returning
endOffsets early which causes the tasks to pause forever and ultimately fail on timeout

[jihoonson]

It looks that this should never happen. Would you elaborate more on when this can happen?

[surekhasaharan]

This happens if no events were passed to kafka tasks, this is the issue from #5666. So checkTaskDuration() removes the groupId taskGroups.remove(groupId); but later checkPendingCompletionTasks() tries to get the groupId in sequenceTaskGroup.remove(generateSequenceName(groupId));

[surekhasaharan]

@jihoonson I tried again, removing the null check, and I cannot reproduce the NPE now with the task resume fix in, but I think, this null check couldn't hurt.

[jihoonson]

@surekhasaharan thanks. Null check is always good, but I'm not sure about returning null in this method. If groupId is never expected to be null, we should throw an exception. Otherwise, this method can return null, but all callers should check the returned sequenceName is null or not. What do you think?

[gianm]

I agree with @jihoonson -- let's fix this in a different patch, since it's really a different bug from the main one we're fixing. The main one being:

If the endOffset is same as startOffset, still let the task resume instead of returning
endOffsets early which causes the tasks to pause forever and ultimately fail on timeout

I raised a new issue for this NPE: #5900

[surekhasaharan]

@jihoonson Agree that all the callers will need to handle null in this case. Will address the possible NPE in separate issue raised by @gianm . Removing the null check here.

[gianm]

@surekhasaharan The test failure seems legit:

testIncrementalHandOff[isIncrementalHandoffSupported = true](io.druid.indexing.kafka.KafkaIndexTaskTest)  Time elapsed: 2.147 sec  <<< FAILURE!
java.lang.AssertionError: expected:<SUCCESS> but was:<FAILED>
	at org.junit.Assert.fail(Assert.java:88)
	at org.junit.Assert.failNotEquals(Assert.java:743)
	at org.junit.Assert.assertEquals(Assert.java:118)
	at org.junit.Assert.assertEquals(Assert.java:144)
	at io.druid.indexing.kafka.KafkaIndexTaskTest.testIncrementalHandOff(KafkaIndexTaskTest.java:517)

[surekhasaharan]

All tests in KafkaIndexTaskTest seem to pass locally.

[gianm]

@surekhasaharan Ah okay cool - I restarted it and it passed.