You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/12/02 22:04:15 UTC
[GitHub] [kafka] lihaosky opened a new pull request, #12944: [KAFKA-14395] add config to configure client supplier
lihaosky opened a new pull request, #12944:
URL: https://github.com/apache/kafka/pull/12944
## Description
This PR is for https://cwiki.apache.org/confluence/display/KAFKA/KIP-884%3A+Add+config+to+configure+KafkaClientSupplier+in+Kafka+Streams. It adds a new `default.client.supplier` config to configure the `KafkaClientSupplier`.
## Test
Unit test
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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
[GitHub] [kafka] lihaosky commented on a diff in pull request #12944: [KAFKA-14395] add config to configure client supplier
Posted by GitBox <gi...@apache.org>.
lihaosky commented on code in PR #12944:
URL: https://github.com/apache/kafka/pull/12944#discussion_r1040472261
##########
streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java:
##########
@@ -1088,6 +1088,28 @@ public void shouldTriggerRecordingOfRocksDBMetricsIfRecordingLevelIsDebug() {
}
}
+ @Test
+ public void shouldGetClientSupplierFromConfigForConstructor() {
+ final StreamsConfig config = new StreamsConfig(props);
+ final StreamsConfig mockConfig = spy(config);
+ when(mockConfig.getKafkaClientSupplier()).thenReturn(supplier);
+
+ new KafkaStreams(getBuilderWithSource().build(), mockConfig);
+ // It's called once in above when mock
+ verify(mockConfig, times(2)).getKafkaClientSupplier();
Review Comment:
Somehow above `when(mockConfig.getKafkaClientSupplier()).thenReturn(supplier);` is counted.
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1727,6 +1738,15 @@ public static Set<String> verifyTopologyOptimizationConfigs(final String config)
return verifiedConfigs;
}
+ /**
+ * Return configured KafkaClientSupplier
+ * @return Configured KafkaClientSupplier
+ */
+ public KafkaClientSupplier getKafkaClientSupplier() {
Review Comment:
Sure. Will update the KIP.
##########
streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java:
##########
@@ -1088,6 +1088,28 @@ public void shouldTriggerRecordingOfRocksDBMetricsIfRecordingLevelIsDebug() {
}
}
+ @Test
+ public void shouldGetClientSupplierFromConfigForConstructor() {
Review Comment:
Sure.
--
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
[GitHub] [kafka] mjsax merged pull request #12944: KAFKA-14395: add config to configure client supplier
Posted by GitBox <gi...@apache.org>.
mjsax merged PR #12944:
URL: https://github.com/apache/kafka/pull/12944
--
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
[GitHub] [kafka] mjsax commented on a diff in pull request #12944: [KAFKA-14395] add config to configure client supplier
Posted by GitBox <gi...@apache.org>.
mjsax commented on code in PR #12944:
URL: https://github.com/apache/kafka/pull/12944#discussion_r1042536401
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1076,7 +1082,12 @@ public class StreamsConfig extends AbstractConfig {
Type.LONG,
null,
Importance.LOW,
- WINDOW_SIZE_MS_DOC);
+ WINDOW_SIZE_MS_DOC)
+ .define(DEFAULT_CLIENT_SUPPLIER_CONFIG,
Review Comment:
nit: we keep the configs grouped by priority and order alphabetically within each group. Can we more this accordingly?
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1727,6 +1738,15 @@ public static Set<String> verifyTopologyOptimizationConfigs(final String config)
return verifiedConfigs;
}
+ /**
+ * Return configured KafkaClientSupplier
+ * @return Configured KafkaClientSupplier
+ */
+ public KafkaClientSupplier getKafkaClientSupplier() {
Review Comment:
Thanks. Can you send a follow up email to the vote thread, calling out the change?
We don't need to re-vote but if anybody would have an objection, we need to give them a chance to race it.
--
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
[GitHub] [kafka] lihaosky commented on a diff in pull request #12944: [KAFKA-14395] add config to configure client supplier
Posted by GitBox <gi...@apache.org>.
lihaosky commented on code in PR #12944:
URL: https://github.com/apache/kafka/pull/12944#discussion_r1042544360
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1076,7 +1082,12 @@ public class StreamsConfig extends AbstractConfig {
Type.LONG,
null,
Importance.LOW,
- WINDOW_SIZE_MS_DOC);
+ WINDOW_SIZE_MS_DOC)
+ .define(DEFAULT_CLIENT_SUPPLIER_CONFIG,
Review Comment:
Sure
--
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
[GitHub] [kafka] mjsax commented on pull request #12944: KAFKA-14395: add config to configure client supplier
Posted by GitBox <gi...@apache.org>.
mjsax commented on PR #12944:
URL: https://github.com/apache/kafka/pull/12944#issuecomment-1351814567
Merged to `trunk`. Thanks for the PR and KIP.
--
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
[GitHub] [kafka] mjsax commented on a diff in pull request #12944: [KAFKA-14395] add config to configure client supplier
Posted by GitBox <gi...@apache.org>.
mjsax commented on code in PR #12944:
URL: https://github.com/apache/kafka/pull/12944#discussion_r1040281808
##########
streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java:
##########
@@ -1088,6 +1088,28 @@ public void shouldTriggerRecordingOfRocksDBMetricsIfRecordingLevelIsDebug() {
}
}
+ @Test
+ public void shouldGetClientSupplierFromConfigForConstructor() {
+ final StreamsConfig config = new StreamsConfig(props);
+ final StreamsConfig mockConfig = spy(config);
+ when(mockConfig.getKafkaClientSupplier()).thenReturn(supplier);
+
+ new KafkaStreams(getBuilderWithSource().build(), mockConfig);
+ // It's called once in above when mock
+ verify(mockConfig, times(2)).getKafkaClientSupplier();
Review Comment:
Why `times(2)` but not `times(1)` ?
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -806,6 +812,11 @@ public class StreamsConfig extends AbstractConfig {
LogAndFailExceptionHandler.class.getName(),
Importance.MEDIUM,
DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
+ .define(DEFAULT_CLIENT_SUPPLIER_CONFIG,
+ Type.CLASS,
+ DefaultKafkaClientSupplier.class.getName(),
+ Importance.MEDIUM,
Review Comment:
I think it should be `LOW` priority.
##########
streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java:
##########
@@ -1088,6 +1088,28 @@ public void shouldTriggerRecordingOfRocksDBMetricsIfRecordingLevelIsDebug() {
}
}
+ @Test
+ public void shouldGetClientSupplierFromConfigForConstructor() {
Review Comment:
Should we also test, that a supplier passed into the constructor overwrites the config?
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1727,6 +1738,15 @@ public static Set<String> verifyTopologyOptimizationConfigs(final String config)
return verifiedConfigs;
}
+ /**
+ * Return configured KafkaClientSupplier
+ * @return Configured KafkaClientSupplier
+ */
+ public KafkaClientSupplier getKafkaClientSupplier() {
Review Comment:
Was this part of the KIP and do we need it? `StreamsConfig` is public API and we cannot just change it.
Also, the client supplier does not implement `Configurable` and thus does not need to be configured.
I am ok with reusing `getConfiguredInstance` but we would need a small change to the KIP.
--
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