Your job is experiencing long durations for the addBatch and latestOffset phases, which suggests there are delays in discovering new data. Can you please try:

• Run OPTIMIZE on the raw Delta table to compact small files and improve metadata handling.
• Enable Delta Optimizations with optimizeWrite and autoCompact to maintain file sizes more effectively.
• Increase minBatchInterval if appropriate for your workload, to allow for larger, less frequent batches.

With availableNow, the job will read from the last known checkpoint offset and scan through all incremental changes. If your Delta table in the raw layer has high data volume or a large number of small files, this can slow down the latestOffset and addBatch phases, as it takes time to retrieve file metadata and determine what’s new. The numFilesOutstanding and numBytesOutstanding metrics indicate there are many small files waiting to be processed, which further contributes to the delay. The latestOffset phase involves checking the latest offsets across multiple files, which can introduce significant latency, particularly if there are many small files due to continuous ingestion from Kinesis. This latency is common in streaming sources that use file-based checkpoints and Delta tables, as it must calculate the cumulative offset of multiple files.

spark.databricks.delta.optimizeWrite.enabled
spark.databricks.delta.autoCompact.enabled

Ref.: https://docs.databricks.com/en/delta/table-properties.html