Revert "Fix unhandled exception in KafkaIO SDF (#37449)"

[johnjcasey]

Reverts #37553

This stops using a flag, but doesn't clean the flag up or cleans up the tests that depend on the flag. While merged, it causes precommit failures for other PRs.

TODO clean up the dangling reference, and the tests: https://github.com/apache/beam/blob/master/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOIT.java#L290, and others

[johnjcasey]

@junaiddshaukat unfortunately it looks like your change caused some precommit failures elsewhere, so I'm reverting it for now. @acrites thanks for finding

[gemini-code-assist]

Summary of Changes

Hello, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request reverts a previous change that introduced soft logging for missing Kafka topics. The intent is to enforce stricter validation by throwing an exception when partition information cannot be retrieved, preventing silent failures in KafkaIO pipelines. This change ensures that misconfigured topics are identified during the pipeline execution rather than being ignored.

Highlights

• Enforcement of Topic Existence: Replaced soft logging warnings with strict state checks when Kafka topic partition information is missing or empty, ensuring configuration issues are caught immediately.
• Test Cleanup: Removed obsolete test cases that validated the previous soft-warning behavior, aligning the test suite with the new strict validation logic.

🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	/gemini review	Performs a code review for the current pull request in its current state.
Pull Request Summary	/gemini summary	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	/gemini help	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here.

Footnotes

1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

[gemini-code-assist]

Code Review

This pull request modifies the Kafka topic partition verification logic in KafkaIO and WatchForKafkaTopicPartitions, transitioning from warning-based logging to strict state checks using checkState. In WatchForKafkaTopicPartitions, the logger was removed entirely in favor of these checks, and associated tests were deleted. However, the implementation in KafkaIO.java contains a logic error where the warning block executes regardless of partition validity when logTopicVerification is enabled, and the removal of the continue statement will lead to a NullPointerException if partitions are missing.

[gemini-code-assist]

The logic in this revert is incorrect. If logTopicVerification is true, the else block executes for every topic, logging a warning even when partitionInfoList is valid. Furthermore, if partitionInfoList is null, the code will proceed to the loop at line 2145 and throw a NullPointerException because the continue statement was removed. The warning and skip logic should only trigger if partitionInfoList is actually null or empty.

Suggested change

	if (logTopicVerification == null || !logTopicVerification) {
	checkState(
	partitionInfoList != null && !partitionInfoList.isEmpty(),
	"Could not find any partitions info for topic %s. Please check Kafka configuration and make sure that provided topics exist.",
	topic);
	} else {
	LOG.warn(
	"Could not find any partitions info for topic {}. Please check Kafka "
	+ "configuration and make sure that the provided topics exist.",
	"Could not find any partitions info for topic {}. Please check Kafka configuration "
	+ "and make sure that the provided topics exist.",
	topic);
	continue;
	}
	if (logTopicVerification == null || !logTopicVerification) {
	checkState(
	partitionInfoList != null && !partitionInfoList.isEmpty(),
	"Could not find any partitions info for topic %s. Please check Kafka configuration and make sure that provided topics exist.",
	topic);
	} else if (partitionInfoList == null || partitionInfoList.isEmpty()) {
	LOG.warn(
	"Could not find any partitions info for topic {}. Please check Kafka configuration "
	+ "and make sure that the provided topics exist.",
	topic);
	continue;
	}

[junaiddshaukat]

@junaiddshaukat unfortunately it looks like your change caused some precommit failures elsewhere, so I'm reverting it for now. @acrites thanks for finding

okay np, should i need to make a new pr for #37449. or just leave it for now