Add checking for new checkpoint

[panhongan]

Fixes #XXXX.

org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisor.CheckpointNotice#handle()

Description

There was NPE when parsed the checkpoint in context.
I checked from the overlord log.

1. Previous task

2023-05-19T02:49:20,653 INFO [qtp1751325469-204] org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisor - Checkpointing [KafkaDataSourceMetadata{Seekable
StreamStartSequenceNumbers=SeekableStreamStartSequenceNumbers{stream='xxxxxxxxx', partitionSequenceNumberMap={16=24076850286}, exclusivePar
titions=[]}}] for taskGroup [16]

2023-05-19T02:49:20,662 WARN [IndexTaskClient-xxxxxxxxx.10-6] org.apache.druid.indexing.common.IndexTaskClient - Exception while sending request
org.apache.druid.java.util.common.IAE: Received 400 Bad Request with body: Can't pause, task is not in a pausable state (state: [PAUSED])
        at org.apache.druid.indexing.common.IndexTaskClient.submitRequest(IndexTaskClient.java:356) ~[druid-indexing-service-0.19.0-iap9.jar:0.19.0-iap9]
        at org.apache.druid.indexing.common.IndexTaskClient.submitRequestWithEmptyContent(IndexTaskClient.java:220) ~[druid-indexing-service-0.19.0-iap9.jar:0.19.0-iap9]
        at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskClient.pause(SeekableStreamIndexTaskClient.java:108) ~[druid-indexing-service-0.19.0-iap9.jar:0.19.0-iap9]
        at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskClient.lambda$pauseAsync$4(SeekableStreamIndexTaskClient.java:339) ~[druid-indexing-service-0.19.0-iap9.jar:0.19.0-iap9]
        at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_222]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_222]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_222]
        at java.lang.Thread.run(Thread.java:748) [?:1.8.0_222]
2023-05-19T02:49:20,668 INFO [KafkaSupervisor-xxxxxxxxx-Worker-7] org.apache.druid.indexing.overlord.RemoteTaskRunner - Shutdown [index_kafka_xxxxxxxxx_0ac74dd2c82397f_knoggfhl] because: [An exception occured while waiting for task [index_kafka_xxxxxxxxx_0ac74dd2c82397f_knoggfhl] to pause: [org.apache.druid.java.util.common.IAE: Received 400 Bad Request with body: Can't pause, task is not in a pausable state (state: [PAUSED])]]

2023-05-19T02:49:20,778 INFO [KafkaSupervisor-xxxxxxxxx-Worker-7] org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisor - All tasks in taskGroup [16] have failed, tasks will be re-created
2023-05-19T02:49:20,778 INFO [KafkaSupervisor-xxxxxxxxx] org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisor - Handled checkpoint notice, new checkpoint is [null] for taskGroup [16]

2. Relay task

2023-05-19T02:58:04,739 ERROR [task-runner-0-priority-0] org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner - Encountered exception while running task.
java.lang.NullPointerException: null
	at com.google.common.base.Preconditions.checkNotNull(Preconditions.java:213) ~[guava-16.0.1.jar:?]
	at org.apache.druid.indexing.seekablestream.SequenceMetadata.<init>(SequenceMetadata.java:80) ~[druid-indexing-service-0.19.0-iap9.jar:0.19.0-iap9]
	at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner.initializeSequences(SeekableStreamIndexTaskRunner.java:329) ~[druid-indexing-service-0.19.0-iap9.jar:0.19.0-iap9]
	at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner.runInternal(SeekableStreamIndexTaskRunner.java:382) ~[druid-indexing-service-0.19.0-iap9.jar:0.19.0-iap9]
	at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner.run(SeekableStreamIndexTaskRunner.java:277) [druid-indexing-service-0.19.0-iap9.jar:0.19.0-iap9]
	at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTask.run(SeekableStreamIndexTask.java:164) [druid-indexing-service-0.19.0-iap9.jar:0.19.0-iap9]
	at org.apache.druid.indexing.overlord.SingleTaskBackgroundRunner$SingleTaskBackgroundRunnerCallable.call(SingleTaskBackgroundRunner.java:421) [druid-indexing-service-0.19.0-iap9.jar:0.19.0-iap9]
	at org.apache.druid.indexing.overlord.SingleTaskBackgroundRunner$SingleTaskBackgroundRunnerCallable.call(SingleTaskBackgroundRunner.java:393) [druid-indexing-service-0.19.0-iap9.jar:0.19.0-iap9]
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_222]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_222]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_222]
	at java.lang.Thread.run(Thread.java:748) [?:1.8.0_222]

And the checkpoint contains null, like:

"context" : {
    "checkpoints" : "{\"0\":{\"16\":24076850286},\"1\":null}",
    "IS_INCREMENTAL_HANDOFF_SUPPORTED" : true,
    "forceTimeChunkLock" : true
  }

Fixed the bug ...

final Map<PartitionIdType, SequenceOffsetType> newCheckpoint = checkpointTaskGroup(taskGroup, false).get();
if (MapUtils.isNotEmpty(newCheckpoint)) {
  taskGroup.addNewCheckpoint(newCheckpoint);
  log.info("Handled checkpoint notice, new checkpoint is [%s] for taskGroup [%s]", newCheckpoint, taskGroupId);
} else {
  log.warn("New checkpoint is null for taskGroup [%s]", taskGroupId);
}

Release note

---

Key changed classes in this PR

• org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisor.CheckpointNotice

---

This PR has:

• been self-reviewed.
  • using the concurrency checklist (Remove this item if the PR doesn't have any relation to concurrency.)
• added documentation for new or modified features or behaviors.
• a release note entry in the PR description.
• added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
• added or updated version, license, or notice information in licenses.yaml
• added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
• added integration tests.
• been tested in a test Druid cluster.

[cryptoe]

@panhongan Have you tested this on a local druid cluster ?

[panhongan]

@panhongan Have you tested this on a local druid cluster ?

Yes, I tested in my local druid cluster.

[AmatyaAvadhanula]

@panhongan thank you for testing this on a cluster. Could you please look at the failing checks as well?

[panhongan]

@panhongan thank you for testing this on a cluster. Could you please look at the failing checks as well?

@AmatyaAvadhanula @cryptoe @pranavbhole
I fixed the issue, can you help review and approve?

[AmatyaAvadhanula]

LGTM @panhongan!