flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-9083) Add async backpressure support to Cassandra Connector
Date Tue, 30 Oct 2018 03:03:00 GMT

    [ https://issues.apache.org/jira/browse/FLINK-9083?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16668058#comment-16668058

ASF GitHub Bot commented on FLINK-9083:

jparkie commented on a change in pull request #6782: [FLINK-9083][Cassandra Connector] Add
async backpressure support to Cassandra Connector
URL: https://github.com/apache/flink/pull/6782#discussion_r229163101

 File path: flink-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSinkBase.java
 @@ -43,70 +48,82 @@
 public abstract class CassandraSinkBase<IN, V> extends RichSinkFunction<IN> implements
CheckpointedFunction {
 	protected final Logger log = LoggerFactory.getLogger(getClass());
-	protected transient Cluster cluster;
-	protected transient Session session;
-	protected transient volatile Throwable exception;
-	protected transient FutureCallback<V> callback;
+	// ------------------------ Default Configurations ------------------------
+	/**
+	 * The default maximum number of concurrent requests. By default, {@code Integer.MAX_VALUE}.
+	 */
+	public static final int DEFAULT_MAX_CONCURRENT_REQUESTS = Integer.MAX_VALUE;
+	/**
+	 * The default timeout duration when acquiring a permit to execute. By default, {@code Long.MAX_VALUE}.
+	 */
+	/**
+	 * The default timeout unit when acquiring a permit to execute. By default, milliseconds.
+	 */
+	// ------------------------- Configuration Fields -------------------------
+	private int maxConcurrentRequests = DEFAULT_MAX_CONCURRENT_REQUESTS;
+	private long maxConcurrentRequestsTimeout = DEFAULT_MAX_CONCURRENT_REQUESTS_TIMEOUT;
+	private TimeUnit maxConcurrentRequestsTimeoutUnit = DEFAULT_MAX_CONCURRENT_REQUESTS_TIMEOUT_UNIT;
 Review comment:
   Yeah. I thought the constructor would explode too much. Furthermore, I was hesitant to
introduce a new constructor that conflicted with the existing if anyone relied on the package
public one.
   However, I like your idea of a config class. I can make one constructor accept that and
have all the old constructors with a `@deprecated` annotation that just delegates to the new

This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:

> Add async backpressure support to Cassandra Connector
> -----------------------------------------------------
>                 Key: FLINK-9083
>                 URL: https://issues.apache.org/jira/browse/FLINK-9083
>             Project: Flink
>          Issue Type: Improvement
>          Components: Cassandra Connector
>            Reporter: Jacob Park
>            Assignee: Jacob Park
>            Priority: Minor
>              Labels: pull-request-available
> As the CassandraSinkBase derivatives utilize async writes, they do not block the task
to introduce any backpressure.
> I am currently using a semaphore to provide backpressure support by blocking at a maximum
concurrent requests limit like how DataStax's Spark Cassandra Connector functions: [https://github.com/datastax/spark-cassandra-connector/blob/v2.0.7/spark-cassandra-connector/src/main/scala/com/datastax/spark/connector/writer/AsyncExecutor.scala#L18]
> This improvement has greatly improved the fault-tolerance of our Cassandra Sink Connector
implementation on Apache Flink in production. I would like to contribute this feature back

This message was sent by Atlassian JIRA

View raw message