fapaul commented on a change in pull request #16023:
URL: https://github.com/apache/flink/pull/16023#discussion_r644716845
##########
File path:
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/common/RMQConnectionConfig.java
##########
@@ -506,16 +532,45 @@ 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(long deliveryTimeout) {
+ Preconditions.checkArgument(
+ deliveryTimeout >= 0, "deliveryTimeout can not be
negative");
+ this.deliveryTimeout = deliveryTimeout;
+ 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 30 seconds.
Review comment:
Nit: update comment if constant is moved.
##########
File path:
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceTest.java
##########
@@ -419,6 +421,40 @@ protected Connection setupConnection() throws Exception {
Mockito.verify(channel, Mockito.times(0)).basicQos(anyInt());
}
+ private static class CallsRealMethodsWithDelay extends CallsRealMethods {
Review comment:
Nit: Can you move this helper after all `@Test` methods?
##########
File path:
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/common/RMQConnectionConfig.java
##########
@@ -506,16 +532,45 @@ 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.
Review comment:
Nit: If you move the constant variable out of this class please also
update the comments.
##########
File path:
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/common/RMQConnectionConfig.java
##########
@@ -29,21 +29,24 @@
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.Optional;
+import java.util.concurrent.TimeUnit;
/**
* Connection Configuration for RMQ. If {@link Builder#setUri(String)} has
been set then {@link
* RMQConnectionConfig#RMQConnectionConfig(String, Integer, Boolean, Boolean,
Integer, Integer,
- * Integer, Integer, Integer)} will be used for initialize the RMQ connection
or {@link
+ * Integer, Integer, Integer, Long)} will be used for initialize the RMQ
connection or {@link
* RMQConnectionConfig#RMQConnectionConfig(String, Integer, String, String,
String, Integer,
- * Boolean, Boolean, Integer, Integer, Integer, Integer, Integer)} will be
used for initialize the
- * RMQ connection
+ * Boolean, Boolean, Integer, Integer, Integer, Integer, Integer, Long)} will
be used for initialize
+ * the RMQ connection
*/
public class RMQConnectionConfig implements Serializable {
private static final long serialVersionUID = 1L;
private static final Logger LOG =
LoggerFactory.getLogger(RMQConnectionConfig.class);
+ private static final long DEFAULT_DELIVERY_TIMEOUT = 30000;
Review comment:
I'd prefer to have this value as part of the `RMQSource` class. This way
this class does not hold any logic and it can be easier reused.
##########
File path:
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceTest.java
##########
@@ -419,6 +421,40 @@ protected Connection setupConnection() throws Exception {
Mockito.verify(channel, Mockito.times(0)).basicQos(anyInt());
}
+ private static class CallsRealMethodsWithDelay extends CallsRealMethods {
+
+ private final long delay;
+
+ public CallsRealMethodsWithDelay(long delay) {
+ this.delay = delay;
+ }
+
+ public Object answer(InvocationOnMock invocation) throws Throwable {
+ Thread.sleep(delay);
+ return super.answer(invocation);
+ }
+ }
+
+ @Test(timeout = 10000L)
+ public void testDeliveryTimeout() throws Exception {
+ source.autoAck = false;
+ // mock delivery delay
+ Mockito.when(source.consumer.nextDelivery())
+ .then(new CallsRealMethodsWithDelay(15000L))
+ .thenThrow(new RuntimeException());
+ Mockito.when(source.consumer.nextDelivery(any(Long.class)))
+ .then(new CallsRealMethodsWithDelay(10L))
+ .thenReturn(null);
+ sourceThread.start();
+ // wait a bit for the source to start
+ Thread.sleep(5);
Review comment:
In general, I'd recommend not relying on time in test cases because it
may take longer on other systems to start the source which may introduce a test
instability.
##########
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:
I agree with @austince's comment.
##########
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:
I think this check should be
```suggestion
Preconditions.checkArgument(
deliveryTimeout != null && deliveryTimeout >= 0,
"deliveryTimeout can not be negative");
```
##########
File path:
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/common/RMQConnectionConfig.java
##########
@@ -132,8 +141,12 @@ private RMQConnectionConfig(
Integer requestedChannelMax,
Integer requestedFrameMax,
Integer requestedHeartbeat,
- Integer prefetchCount) {
+ Integer prefetchCount,
+ Long deliveryTimeout) {
Preconditions.checkNotNull(uri, "Uri can not be null");
+ Preconditions.checkArgument(
+ deliveryTimeout == null || deliveryTimeout >= 0,
+ "deliveryTimeout can not be negative");
Review comment:
Same comment as above.
--
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
