NotificationsYou must be signed in to change notification settings
Fork14.8k
Star31.6k

Commitd785957

authored

MINOR: code cleanup in InternalTopicManager (#21165)

- Avoid double-brace initialization- Fix timeout error handling, by pass in correct list of topic names- Remove unnecessary codeReviewers: Vincent Potuček (@Pankraz76), Lucas Brutschy <lbrutschy@confluent.io>

1 parenta2eae88 commitd785957Copy full SHA for d785957

File tree

1 file changed

+13

-14

lines changed

streams/src/main/java/org/apache/kafka/streams/processor/internals
- InternalTopicManager.java

1 file changed

+13

-14

lines changed

`‎streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java‎`

Lines changed: 13 additions & 14 deletions

Original file line number	Diff line number	Diff line change
`@@ -194,18 +194,19 @@ public ValidationResult validate(final Map<String, InternalTopicConfig> topicCon`
`194`	`194`	`(streamsSide,brokerSide) ->validateCleanupPolicy(validationResult,streamsSide,brokerSide)`
`195`	`195`	`);`
`196`	`196`	`}`
`197`		`-`
	`197`	`+`
	`198`	`+finalSet<String>topicsStillToValidate =newHashSet<>();`
	`199`	`+topicsStillToValidate.addAll(topicDescriptionsStillToValidate);`
	`200`	`+topicsStillToValidate.addAll(topicConfigsStillToValidate);`
	`201`	`+`
`198`	`202`	`maybeThrowTimeout(newTimeoutContext(`
`199`		`-newHashSet<String>() {{`
`200`		`-addAll(topicDescriptionsStillToValidate);`
`201`		`-addAll(topicConfigsStillToValidate);`
`202`		`- }},`
`203`		`-deadline,`
`204`		`-"Validation timeout",`
`205`		`-String.format("Could not validate internal topics within %d milliseconds. " +`
`206`		`-"This can happen if the Kafka cluster is temporarily not available.",retryTimeoutMs),`
`207`		`-null`
`208`		`- ));`
	`203`	`+topicsStillToValidate,`
	`204`	`+deadline,`
	`205`	`+"Validation timeout",`
	`206`	`+String.format("Could not validate internal topics within %d milliseconds. " +`
	`207`	`+"This can happen if the Kafka cluster is temporarily not available.",retryTimeoutMs),`
	`208`	`+null`
	`209`	`+ ));`
`209`	`210`
`210`	`211`	`if (!descriptionsForTopic.isEmpty() \|\| !configsForTopic.isEmpty()) {`
`211`	`212`	`Utils.sleep(100);`
`@@ -497,7 +498,7 @@ public Set<String> makeReady(final Map<String, InternalTopicConfig> topics) {`
`497`	`498`	`}`
`498`	`499`	`if (!topicsNotReady.isEmpty()) {`
`499`	`500`	`maybeThrowTimeout(newTimeoutContext(`
`500`		-Collections.singleton("makeReadyCheck"),// dummy collection just to trigger if `topicsNotReady` is non-empty
	`501`	`+topicsNotReady,`
`501`	`502`	`deadlineMs,`
`502`	`503`	`"MakeReady timeout",`
`503`	`504`	`String.format("Could not create topics within %d milliseconds. This can happen if the Kafka cluster is temporarily not available.",retryTimeoutMs),`
`@@ -613,8 +614,6 @@ private Set<String> createTopics(final Set<NewTopic> topicsToCreate,`
`613`	`614`	`deadlineMs -time.milliseconds()`
`614`	`615`	`);`
`615`	`616`	`Utils.sleep(retryBackOffMs);`
`616`		`- }else {`
`617`		`-continue;`
`618`	`617`	`}`
`619`	`618`	`}`
`620`	`619`	`returncreatedTopics;`

0 commit comments

Comments

(0)

Movatterモバイル変換

Navigation Menu

Search code, repositories, users, issues, pull requests...

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Commitd785957

File tree

1 file changed

1 file changed

`‎streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java‎`

0 commit comments