chia7712 commented on code in PR #17666:
URL: https://github.com/apache/kafka/pull/17666#discussion_r1826531936
##########
test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterInstance.java:
##########
@@ -224,5 +232,46 @@ default void waitAcls(AclBindingFilter filter,
Collection<AccessControlEntry> en
}, "expected acls: " + entries + ", actual acls: " +
actualEntries.get());
}
}
+
+
//---------------------------[producer/consumer]---------------------------//
+
+ default <K, V> Producer<K, V> producer(Map<String, Object> overrides,
+ Serializer<K> keySerializer,
+ Serializer<V> valueSerializer
+ ) {
+ Properties props = new Properties();
+ props.putAll(overrides);
+ props.putIfAbsent(ProducerConfig.ACKS_CONFIG, "-1");
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
+ props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
keySerializer.getClass().getName());
+ props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
valueSerializer.getClass().getName());
+ return new KafkaProducer<>(props);
+ }
+
+ default <K, V> Producer<K, V> producer(Serializer<K> keySerializer,
+ Serializer<V> valueSerializer
+ ) {
+ return producer(Collections.emptyMap(), keySerializer,
valueSerializer);
+ }
+
+ default <K, V> Consumer<K, V> consumer(Map<String, Object> overrides,
+ Deserializer<K> keyDeserializer,
+ Deserializer<V> valueDeserializer
+ ) {
+ Properties props = new Properties();
+ props.putAll(overrides);
Review Comment:
we should call this later. otherwise, it is not "overrides"
##########
test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/RaftClusterInvocationContext.java:
##########
@@ -175,8 +175,10 @@ public KafkaClusterTestKit getUnderlying() {
}
@Override
- public Admin createAdminClient(Properties configOverrides) {
- Admin admin =
Admin.create(clusterTestKit.newClientPropertiesBuilder(configOverrides).build());
+ public Admin admin(Map<String, Object> overrides) {
+ Properties props = new Properties();
+ props.putAll(overrides);
+ Admin admin =
Admin.create(clusterTestKit.newClientPropertiesBuilder(props).build());
admins.add(admin);
Review Comment:
could you please remove this collection and move this implementation to
interface?
##########
test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterInstance.java:
##########
@@ -147,10 +155,10 @@ default <T> T getUnderlying(Class<T> asClass) {
return asClass.cast(getUnderlying());
}
- Admin createAdminClient(Properties configOverrides);
+ Admin admin(Map<String, Object> overrides);
Review Comment:
could you please put them (admin, consumer and producer) together?
##########
test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterInstance.java:
##########
@@ -147,10 +155,10 @@ default <T> T getUnderlying(Class<T> asClass) {
return asClass.cast(getUnderlying());
}
- Admin createAdminClient(Properties configOverrides);
+ Admin admin(Map<String, Object> overrides);
- default Admin createAdminClient() {
- return createAdminClient(new Properties());
+ default Admin admin() {
+ return admin(Collections.emptyMap());
Review Comment:
`Map.of`
##########
test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterInstance.java:
##########
@@ -224,5 +232,46 @@ default void waitAcls(AclBindingFilter filter,
Collection<AccessControlEntry> en
}, "expected acls: " + entries + ", actual acls: " +
actualEntries.get());
}
}
+
+
//---------------------------[producer/consumer]---------------------------//
+
+ default <K, V> Producer<K, V> producer(Map<String, Object> overrides,
+ Serializer<K> keySerializer,
+ Serializer<V> valueSerializer
+ ) {
+ Properties props = new Properties();
+ props.putAll(overrides);
+ props.putIfAbsent(ProducerConfig.ACKS_CONFIG, "-1");
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
+ props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
keySerializer.getClass().getName());
+ props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
valueSerializer.getClass().getName());
+ return new KafkaProducer<>(props);
+ }
+
+ default <K, V> Producer<K, V> producer(Serializer<K> keySerializer,
+ Serializer<V> valueSerializer
+ ) {
+ return producer(Collections.emptyMap(), keySerializer,
valueSerializer);
+ }
+
+ default <K, V> Consumer<K, V> consumer(Map<String, Object> overrides,
+ Deserializer<K> keyDeserializer,
+ Deserializer<V> valueDeserializer
+ ) {
+ Properties props = new Properties();
+ props.putAll(overrides);
+ props.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ props.putIfAbsent(ConsumerConfig.GROUP_ID_CONFIG, "group");
+ props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
+ props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
keyDeserializer.getClass().getName());
Review Comment:
maybe this helper method doesn't need to have `keyDeserializer`, as caller
can cast the returned type manually.
##########
test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterInstance.java:
##########
@@ -224,5 +232,46 @@ default void waitAcls(AclBindingFilter filter,
Collection<AccessControlEntry> en
}, "expected acls: " + entries + ", actual acls: " +
actualEntries.get());
}
}
+
+
//---------------------------[producer/consumer]---------------------------//
+
+ default <K, V> Producer<K, V> producer(Map<String, Object> overrides,
+ Serializer<K> keySerializer,
+ Serializer<V> valueSerializer
+ ) {
+ Properties props = new Properties();
+ props.putAll(overrides);
+ props.putIfAbsent(ProducerConfig.ACKS_CONFIG, "-1");
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
+ props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
keySerializer.getClass().getName());
+ props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
valueSerializer.getClass().getName());
+ return new KafkaProducer<>(props);
+ }
+
+ default <K, V> Producer<K, V> producer(Serializer<K> keySerializer,
+ Serializer<V> valueSerializer
+ ) {
+ return producer(Collections.emptyMap(), keySerializer,
valueSerializer);
+ }
+
+ default <K, V> Consumer<K, V> consumer(Map<String, Object> overrides,
+ Deserializer<K> keyDeserializer,
+ Deserializer<V> valueDeserializer
+ ) {
+ Properties props = new Properties();
+ props.putAll(overrides);
+ props.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ props.putIfAbsent(ConsumerConfig.GROUP_ID_CONFIG, "group");
Review Comment:
maybe we should create random string to be the group
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
