[SUPPORT] Streaming improvements

**_Tips before filing an issue_**

- Have you gone through our [FAQs](https://hudi.apache.org/learn/faq/)?

- Join the mailing list to engage in conversations and get faster support at dev-subscribe@hudi.apache.org.

- If you have triaged this as a bug, then file an [issue](https://issues.apache.org/jira/projects/HUDI/issues) directly.

**Describe the problem you faced**

As of now - the Hudi streaming source has limited features compared to well-known Sources, such as Apache Kafka.
We needed such functionality and managed to add it and run it for > 6 months with no problems.
The list of features we have among the current master:
1. Support of TriggerAvailableNow
2. Support for Admission control(similar to Kafka maxTriggerDelay/minNumberRows/etc) - allows you to limit the number of rows processed in a single batch(`conf` value setting) - for this, we read the commit line and pack commits given number of changed rows inside - ofc this will lead to final number be less than customer set, because some rows may have updates.
3. Metrics about this logic(like batch delayed/started/initial run).
4. Metrics about backlog - such as how many rows are in backlog(logic from above) / batches to be run(based on rows/records per batch).
5. custom start/end time - currently only `start` can be defined.

Given this - would the Hudi team be interested in this? I can create the Pull Request.

**To Reproduce**

Steps to reproduce the behavior:

1.
6.
7.
8.

**Expected behavior**

A clear and concise description of what you expected to happen.

**Environment Description**

* Hudi version :

* Spark version :

* Hive version :

* Hadoop version :

* Storage (HDFS/S3/GCS..) :

* Running on Docker? (yes/no) :

**Additional context**

Add any other context about the problem here.

**Stacktrace**

```Add the stacktrace of the error.```

The issue does not name files, tests, or an entry point. Start by locating Hudi's streaming source and reviewing its current trigger and start-position behavior; compare it with the requested TriggerAvailableNow, admission-control, backlog metrics, and custom start/end-time capabilities. Done means the agreed feature scope is implemented and covered by verification for each requested behavior.

Written by the indexing model from the issue text.