dannycranmer commented on code in PR #20245: URL: https://github.com/apache/flink/pull/20245#discussion_r924174639
########## flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/strategy/RequestInfo.java: ########## @@ -0,0 +1,65 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.connector.base.sink.writer.strategy; + +import org.apache.flink.annotation.PublicEvolving; + +/** Dataclass to encapsulate information about starting requests. */ +@PublicEvolving +public class RequestInfo { + private final int batchSize; + private final long requestStartTime; Review Comment: It is not clear here what the unit is. Suggest you use `Instant` instead of long ########## flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriter.java: ########## @@ -344,69 +351,69 @@ public void write(InputT element, Context context) throws IOException, Interrupt * </ul> */ private void nonBlockingFlush() throws InterruptedException { - while (!isInFlightRequestOrMessageLimitExceeded() + while (!rateLimitingStrategy.shouldBlock(createRequestInfo()) && (bufferedRequestEntries.size() >= getNextBatchSizeLimit() || bufferedRequestEntriesTotalSizeInBytes >= maxBatchSizeInBytes)) { flush(); } } - /** - * Determines if the sink should block and complete existing in flight requests before it may - * prudently create any new ones. This is exactly determined by if the number of requests - * currently in flight exceeds the maximum supported by the sink OR if the number of in flight - * messages exceeds the maximum determined to be appropriate by the rate limiting strategy. - */ - private boolean isInFlightRequestOrMessageLimitExceeded() { - return inFlightRequestsCount >= maxInFlightRequests - || inFlightMessages >= rateLimitingStrategy.getRateLimit(); + private RequestInfo createRequestInfo() { + int batchSize = getNextBatchSize(); + long requestStartTime = System.currentTimeMillis(); + return RequestInfo.builder() + .setBatchSize(batchSize) + .setRequestStartTime(requestStartTime) + .build(); } /** * Persists buffered RequestsEntries into the destination by invoking {@code * submitRequestEntries} with batches according to the user specified buffering hints. * - * <p>The method blocks if too many async requests are in flight. + * <p>The method checks with the {@code rateLimitingStrategy} to see if it should block the + * request. */ private void flush() throws InterruptedException { - while (isInFlightRequestOrMessageLimitExceeded()) { + RequestInfo requestInfo = createRequestInfo(); + while (rateLimitingStrategy.shouldBlock(requestInfo)) { mailboxExecutor.yield(); + requestInfo = createRequestInfo(); } - List<RequestEntryT> batch = createNextAvailableBatch(); - int batchSize = batch.size(); - + List<RequestEntryT> batch = createNextAvailableBatch(requestInfo); if (batch.size() == 0) { return; } - long timestampOfRequest = System.currentTimeMillis(); - Consumer<List<RequestEntryT>> requestResult = + int batchSize = requestInfo.getBatchSize(); Review Comment: At this point is `requestInfo.getBatchSize()` guaranteed to equal `batch.size()`? I am guessing not from line 385. In which case is this the right one to use? ########## flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/strategy/RateLimitingStrategy.java: ########## @@ -0,0 +1,59 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.connector.base.sink.writer.strategy; + +import org.apache.flink.annotation.PublicEvolving; + +/** + * Controls the rate of requests being made in the {@code AsyncSinkWriter}. Review Comment: nit: It is a smell to know about `AsyncSinkWriter` here ########## flink-end-to-end-tests/flink-end-to-end-tests-aws-kinesis-firehose/src/test/resources/send-orders.sql: ########## @@ -29,7 +29,7 @@ CREATE TABLE orders ( 'aws.credentials.basic.secretkey' = 'secretAccessKey', 'aws.trust.all.certificates' = 'true', 'sink.http-client.protocol.version' = 'HTTP1_1', - 'sink.batch.max-size' = '1', + 'sink.batch.max-size' = '2', Review Comment: Why did you need to change this? ########## flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterTest.java: ########## @@ -19,6 +19,8 @@ import org.apache.flink.api.common.operators.MailboxExecutor; import org.apache.flink.api.connector.sink2.Sink; +import org.apache.flink.connector.base.sink.writer.strategy.AIMDScalingStrategy; +import org.apache.flink.connector.base.sink.writer.strategy.CongestionControlRateLimitingStrategy; import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService; import org.junit.jupiter.api.BeforeEach; Review Comment: There are no new tests in here. Are these changes adequately captured by the existing tests? I suppose this is a non-functional change for AIMD -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
