You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "k-apol (via GitHub)" <gi...@apache.org> on 2026/03/29 21:20:15 UTC
[PR] KAFKA-10357: Add explicit internal topic initialization for Kafka Str… [kafka]
k-apol opened a new pull request, #21892:
URL: https://github.com/apache/kafka/pull/21892
Implements KIP-698 to prevent silent data loss when internal topics (repartition/changelog) are accidentally deleted.
New public API:
- KafkaStreams#init() and init(InitParameters) to explicitly create and validate internal topics before starting the application
- InitParameters with fluent builder (following CloseOptions pattern): timeout(), withTimeout(), enableSetupInternalTopicsIfIncomplete()
- StreamsConfig 'internal.topics.setup' config: 'automatic' (default, backward compatible) or 'manual' (requires init(), throws on missing topics during rebalance)
New exceptions (already merged via #19988):
- MissingInternalTopicsException
- MisconfiguredInternalTopicException
- InternalTopicsAlreadySetupException
Behavioral changes:
- In manual mode, makeReady() throws MissingInternalTopicsException when topics need creating during rebalance (instead of silently recreating them)
- init() validates state (CREATED only), source topics, internal topic existence and configuration
- INTERNAL_TOPIC_SETUP_CONFIG forwarded to consumer configs for the partition assignor
Reviewers: Matthias J. Sax <ma...@confluent.io>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10357: Add explicit internal topic initialization for Kafka Str… [kafka]
Posted by "zheguang (via GitHub)" <gi...@apache.org>.
zheguang commented on code in PR #21892:
URL: https://github.com/apache/kafka/pull/21892#discussion_r3359413799
##########
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java:
##########
@@ -297,6 +305,132 @@ public boolean isValidTransition(final State newState) {
private final Object stateLock = new Object();
protected volatile State state = State.CREATED;
+ /**
+ * Initializes broker-side state for this Kafka Streams application.
+ * <p>
+ * Kafka Streams creates internal topics on the broker for fault-tolerance and repartitioning.
+ * This method validates and optionally creates those internal topics before starting the application.
+ *
+ * @throws MissingSourceTopicException if a source topic is missing
+ * @throws MissingInternalTopicsException if some but not all of the internal topics are missing
+ * @throws MisconfiguredInternalTopicException if an internal topic is misconfigured
+ * @throws InternalTopicsAlreadySetupException if all internal topics are already setup
+ */
+ public void init() {
+ this.doInit(new InitParameters());
+ }
+
+ /**
+ * Initializes broker-side state for this Kafka Streams application.
+ * <p>
+ * Kafka Streams creates internal topics on the broker for fault-tolerance and repartitioning.
+ * This method validates and optionally creates those internal topics before starting the application.
+ *
+ * @param initParameters parameters controlling initialization behavior
+ * @throws MissingSourceTopicException if a source topic is missing
+ * @throws MissingInternalTopicsException if some but not all of the internal topics are missing
+ * and the parameters do not specify to set them up
+ * @throws MisconfiguredInternalTopicException if an internal topic is misconfigured
+ * @throws InternalTopicsAlreadySetupException if all internal topics are already setup
+ * @throws TimeoutException if initialization exceeds the configured timeout
+ */
+ public void init(final InitParameters initParameters) {
Review Comment:
Looks like here is missing implementation for two other public interfaces proposed in the [KIP](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=165226988#KIP698:AddExplicitUserInitializationofBrokersideStatetoKafkaStreams-PublicInterfaces):
```java
public void init(final Duration timeout);
public void init(final InitParameters initParameters, final Duration timeout);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10357: Add explicit internal topic initialization for Kafka Str… [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21892:
URL: https://github.com/apache/kafka/pull/21892#issuecomment-4190236634
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org