cmick commented on a change in pull request #16023:
URL: https://github.com/apache/flink/pull/16023#discussion_r645573846
##########
File path:
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/common/RMQConnectionConfig.java
##########
@@ -90,12 +94,16 @@ private RMQConnectionConfig(
Integer requestedChannelMax,
Integer requestedFrameMax,
Integer requestedHeartbeat,
- Integer prefetchCount) {
+ Integer prefetchCount,
+ Long deliveryTimeout) {
Preconditions.checkNotNull(host, "host can not be null");
Preconditions.checkNotNull(port, "port can not be null");
Preconditions.checkNotNull(virtualHost, "virtualHost can not be null");
Preconditions.checkNotNull(username, "username can not be null");
Preconditions.checkNotNull(password, "password can not be null");
+ Preconditions.checkArgument(
+ deliveryTimeout == null || deliveryTimeout >= 0,
+ "deliveryTimeout can not be negative");
Review comment:
The check is correct. The value should either be not set at all (so
`null` is fine, default will be used here) or set to non-negative number.
It could be refactored a bit, so that the config would `long` instead of
nullable `Long` and the default (if not provided) would be set by the builder.
But I would have doubts if this is a good idea. Currently the builder is just a
helper to build the config instance, with no internal logic.
What's your opinion? Or maybe you have some other ideas how to make it more
clear?
##########
File path:
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/common/RMQConnectionConfig.java
##########
@@ -506,16 +532,44 @@ public Builder setPrefetchCount(int prefetchCount) {
return this;
}
+ /**
+ * Enables setting the next message delivery timeout in the queueing
consumer. Only
+ * applicable to the {@link RMQSource}. Set to 0 for unlimited (the
consumer will be blocked
+ * until an element becomes available). Default is 30000.
+ *
+ * @param deliveryTimeout maximum wait time, in milliseconds, for the
next message delivery
+ * @return the Builder
+ */
+ public Builder setDeliveryTimeout(int deliveryTimeout) {
+ this.deliveryTimeout = deliveryTimeout;
Review comment:
@austince Sounds good, but would you agree that `return
setDeliveryTimeout(unit.toMillis(deliveryTimeout));` in the other overloaded
method will also do the job? (added in a384aa9)
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
