[controller] Fix mismatch between hybrid version partition count and real-time partition count

[sushantmane]

Ensure real-time topic partition count matches hybrid version partition count

This fix addresses an issue where the real-time topic partition count did not align with the hybrid version
partition count, causing errors during hybrid store ingestion. The issue occurred in the following scenario:

1. Create a store with 1 partition.
2. Perform a batch push, creating a batch version with 1 partition.
3. Update the store to 3 partitions and convert it to a hybrid store.
4. Start real-time writes using push type STREAM.
5. Perform a full push to create a hybrid version with 3 partitions. This push fails because, after the topic
   switch, real-time consumers cannot find partitions 2 and 3 due to the real-time topic having only 1 partition.

Root Cause:

• In step 4, if the real-time topic did not exist, it was created with a partition count derived from the largest
  existing version (batch version with 1 partition), which lead to mismatch.

Solution in this PR:

• STREAM push type is now disallowed if there is no online hybrid version.
• If an online hybrid version exists, it ensures the real-time topic partition count matches the hybrid version
  partition count.
• The requestTopicForPushing method no longer creates a real-time topic if it does not already exist with STREAM PushType.
• For newly created hybrid version we create RT topic when we create version instead of creating it during topic switch. The logic for RT topic creation during TS has been kept around for backwards compatibility but it will be removed once release with new change is rolled out everywhere.

Misc:

• Refactor CreationVersion::requestTopicForPushing to make it easy for unit testing
• Added similar checks for incremental push job type
• Fix flaky blob transfer flaky tests
• Stop creating real-time topics for region system stores like meta and PS3 in parent region
• Logging changes around TS to make it easy to search

How was this PR tested?

WIP

Does this PR introduce any user-facing changes?

• No. You can skip the rest of this section.
• Yes. Make sure to explain your proposed changes and call out the behavior change.

[sushantmane]

Moving to draft as I would like to move topic creation in addVersion and see how that works out

[arjun4084346]

I agree with the solution "STREAM push type is now disallowed if there is no online hybrid version." However, I think, instead of returning error, we should create a new version (that would be of right partition count) and then let the push start.
If we do not auto create the version, we can at least tell the user to create a version, if we find store's partition count different from the VT's partition count.

If we look carefully, changing partition count without creating a version (even for batch stores) is similar to changing partition for a hybrid store; even for the batch stores, because batch stores can be changed to hybrid

I am skeptical about "The requestTopicForPushing method no longer creates a real-time topic if it does not already exist."; I believe in most of the cases, this is a benign operation and we should not prevent creation of the RT.

[sushantmane]

I agree with the solution "STREAM push type is now disallowed if there is no online hybrid version." However, I think, instead of returning error, we should create a new version (that would be of right partition count) and then let the push start. If we do not auto create the version, we can at least tell the user to create a version, if we find store's partition count different from the VT's partition count.

If we look carefully, changing partition count without creating a version (even for batch stores) is similar to changing partition for a hybrid store; even for the batch stores, because batch stores can be changed to hybrid

I am skeptical about "The requestTopicForPushing method no longer creates a real-time topic if it does not already exist."; I believe in most of the cases, this is a benign operation and we should not prevent creation of the RT.

we should create a new version (that would be of right partition count)

I think this is not safe in some cases, for example, user wants to create a new version but keep old data.

If we look carefully, changing partition count without creating a version (even for batch stores) is similar to changing partition for a hybrid store; even for the batch stores, because batch stores can be changed to hybrid

I'm don't really follow the rest of the comment. Could you please elaborate? Thanks

changing partition for a hybrid store

We do not allow this as of now

I believe in most cases, this is a harmless operation, and we should not prevent the creation of the RT.

I disagree that this is a harmless operation. The case mentioned in the commit message is one example why this is not a good idea. Moreover, we should tie hybrid version materialization to RT creation for better alignment and not let RT creation happen at every other place IMHO

[sixpluszero]

Did not read this code path before, but I am confused by the method implementation, as it seems like it requires the retrieved hybrid version to be terminal status, which I don't think is really needed?

[sixpluszero]

Just to clarify that by implementation, inc push is only pushed to one of the above topics based on VPJ config. But I think more checking is always good. nit: can consider wrapping the lines into a small check method and reuse for RT and sepRT.