You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/03/17 17:19:23 UTC
[GitHub] [flink-web] CrynetLogistics opened a new pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
CrynetLogistics opened a new pull request #517:
URL: https://github.com/apache/flink-web/pull/517
… Base, new for 1.15
Documentation for the Async Base Sink FLINK-24041.
This work was originally planned to go into the documentation in the apache/flink repository, but advice from committers has been a blog post is more appropriate for this type of content. Please see FLINK-24370 and https://github.com/apache/flink/pull/17349.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] infoverload commented on a change in pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
infoverload commented on a change in pull request #517:
URL: https://github.com/apache/flink-web/pull/517#discussion_r838139849
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
Review comment:
```suggestion
title: "The Generic Asynchronous Base Sink"
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] MartijnVisser commented on pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
MartijnVisser commented on pull request #517:
URL: https://github.com/apache/flink-web/pull/517#issuecomment-1072185382
@infoverload Will you do a review on this?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] infoverload commented on a change in pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
infoverload commented on a change in pull request #517:
URL: https://github.com/apache/flink-web/pull/517#discussion_r838146254
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
+
+New for [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the Async Base Sink - an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
+
+This common abstraction will reduce the effort required to maintain all these individual sinks, with bugfixes and improvements to the sink core benefiting all implementations that extend it.
+
+**Attention** The sink is designed to participate in checkpointing to provide at-least once semantics, but it is limited to destinations that provide a client that supports async requests.
+
+The design of the sink focuses on extensibility and a broad support of destinations. The core of the sink is kept generic and free of any connector specific dependencies.
+
+
+{% toc %}
+
+# Dependency
+To use this base sink, add the following dependency to your project:
+
+```xml
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.version}</version>
+</dependency>
+```
+
+
+
+# Public Interfaces
+
+## Generic Types
+
+`<InputT>` – type of elements in a DataStream that should be passed to the sink
+
+`<RequestEntryT>` – type of a payload containing the element and additional metadata that is required to submit a single element to the destination
+
+
+## Element Converter Interface
+```java
+public interface ElementConverter<InputT, RequestEntryT> extends Serializable {
+ RequestEntryT apply(InputT element, SinkWriter.Context context);
+}
+```
+Concrete sink implementers should provide a way to convert from an element in the DataStream to the payload type that contains all the additional metadata required to submit that element to the destination by the sink.
+Ideally this would be hidden from the end user as it allows concrete sink implementers to adapt to changes in the destination api without breaking end user code.
Review comment:
```suggestion
## ElementConverter Interface
```java
public interface ElementConverter<InputT, RequestEntryT> extends Serializable {
RequestEntryT apply(InputT element, SinkWriter.Context context);
}
```
Classes that implement the concrete sink should provide a way to convert from an element in the DataStream to the payload type that contains all the additional metadata required to submit that element to the destination by the sink. Ideally, this would be hidden from the end user since it allows concrete sink implementers to adapt to changes in the destination API without breaking end user code.
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] infoverload commented on a change in pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
infoverload commented on a change in pull request #517:
URL: https://github.com/apache/flink-web/pull/517#discussion_r838146254
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
+
+New for [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the Async Base Sink - an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
+
+This common abstraction will reduce the effort required to maintain all these individual sinks, with bugfixes and improvements to the sink core benefiting all implementations that extend it.
+
+**Attention** The sink is designed to participate in checkpointing to provide at-least once semantics, but it is limited to destinations that provide a client that supports async requests.
+
+The design of the sink focuses on extensibility and a broad support of destinations. The core of the sink is kept generic and free of any connector specific dependencies.
+
+
+{% toc %}
+
+# Dependency
+To use this base sink, add the following dependency to your project:
+
+```xml
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.version}</version>
+</dependency>
+```
+
+
+
+# Public Interfaces
+
+## Generic Types
+
+`<InputT>` – type of elements in a DataStream that should be passed to the sink
+
+`<RequestEntryT>` – type of a payload containing the element and additional metadata that is required to submit a single element to the destination
+
+
+## Element Converter Interface
+```java
+public interface ElementConverter<InputT, RequestEntryT> extends Serializable {
+ RequestEntryT apply(InputT element, SinkWriter.Context context);
+}
+```
+Concrete sink implementers should provide a way to convert from an element in the DataStream to the payload type that contains all the additional metadata required to submit that element to the destination by the sink.
+Ideally this would be hidden from the end user as it allows concrete sink implementers to adapt to changes in the destination api without breaking end user code.
Review comment:
```suggestion
## ElementConverter Interface
```java
public interface ElementConverter<InputT, RequestEntryT> extends Serializable {
RequestEntryT apply(InputT element, SinkWriter.Context context);
}
```
Classes that implement the concrete sink should provide a way to convert from an element in the DataStream to the payload type that contains all the additional metadata required to submit that element to the destination by the sink. Ideally, this would be hidden from the end user since it allows concrete sink implementers to adapt to changes in the destination API without breaking end user code.
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] MartijnVisser commented on pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
MartijnVisser commented on pull request #517:
URL: https://github.com/apache/flink-web/pull/517#issuecomment-1072185382
@infoverload Will you do a review on this?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] infoverload commented on a change in pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
infoverload commented on a change in pull request #517:
URL: https://github.com/apache/flink-web/pull/517#discussion_r838148790
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
+
+New for [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the Async Base Sink - an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
+
+This common abstraction will reduce the effort required to maintain all these individual sinks, with bugfixes and improvements to the sink core benefiting all implementations that extend it.
+
+**Attention** The sink is designed to participate in checkpointing to provide at-least once semantics, but it is limited to destinations that provide a client that supports async requests.
+
+The design of the sink focuses on extensibility and a broad support of destinations. The core of the sink is kept generic and free of any connector specific dependencies.
+
+
+{% toc %}
+
+# Dependency
+To use this base sink, add the following dependency to your project:
+
+```xml
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.version}</version>
+</dependency>
+```
+
+
+
+# Public Interfaces
+
+## Generic Types
+
+`<InputT>` – type of elements in a DataStream that should be passed to the sink
+
+`<RequestEntryT>` – type of a payload containing the element and additional metadata that is required to submit a single element to the destination
+
+
+## Element Converter Interface
+```java
+public interface ElementConverter<InputT, RequestEntryT> extends Serializable {
+ RequestEntryT apply(InputT element, SinkWriter.Context context);
+}
+```
+Concrete sink implementers should provide a way to convert from an element in the DataStream to the payload type that contains all the additional metadata required to submit that element to the destination by the sink.
+Ideally this would be hidden from the end user as it allows concrete sink implementers to adapt to changes in the destination api without breaking end user code.
+
+
+## Sink Writer Interface
+
+```java
+public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable>
+ implements SinkWriter<InputT, Void, Collection<RequestEntryT>> {
+ // ...
+ protected abstract void submitRequestEntries
+ (List<RequestEntryT> requestEntries, Consumer<List<RequestEntryT>> requestResult);
+ // ...
+}
+```
+
+This method should specify how a list of elements from the datastream may be persisted into the destination. Sink implementers of various datastore and data processing vendors may use their own clients in connecting to and persisting the requestEntries received by this method.
+
+Should any elements fail to be persisted, they should be requeued back in the buffer for retry using `requestResult.accept(...list of failed entries...)`. However, retrying any element that is known to be faulty and consistently failing, will result in that element being requeued forever, therefore a sensible strategy for determining what should be retried is highly recommended.
+
+```java
+public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable>
+ implements SinkWriter<InputT, Void, Collection<RequestEntryT>> {
+ // ...
+ protected abstract long getSizeInBytes(RequestEntryT requestEntry);
+ // ...
+}
+```
+The generic sink has a concept of size of elements in the buffer. This allows users to specify a byte size threshold beyond which elements will be flushed. However the sink implementer is best positioned to determine what is most sensible measure of size for each `RequestEntryT` is.
+
+
+```java
+public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable>
+ implements SinkWriter<InputT, Void, Collection<RequestEntryT>> {
+ // ...
+ public AsyncSinkWriter(
+ ElementConverter<InputT, RequestEntryT> elementConverter,
+ Sink.InitContext context,
+ int maxBatchSize,
+ int maxInFlightRequests,
+ int maxBufferedRequests,
+ long flushOnBufferSizeInBytes,
+ long maxTimeInBufferMS,
+ long maxRecordSizeInBytes) { /* ... */ }
+ // ...
+}
+```
+
+## Sink Interface
+
+```java
+public class MySink<InputT> extends AsyncSinkBase<InputT, RequestEntryT> {
+ // ...
+ @Override
+ public SinkWriter<InputT, Void, Collection<PutRecordsRequestEntry>> createWriter
+ (InitContext context, List<Collection<PutRecordsRequestEntry>> states) {
+ return new MySinkWriter(context);
+ }
+ // ...
+}
+```
+Sink implementers extending this will need to return their own extension of the `AsyncSinkWriter` from `createWriter()` inside their own implementation of `AsyncSinkBase`.
+
+Currently the [Kinesis Data Streams sink](https://github.com/apache/flink/tree/master/flink-connectors/flink-connector-aws-kinesis-streams) and [Kinesis Data Firehose sink](https://github.com/apache/flink/tree/master/flink-connectors/flink-connector-aws-kinesis-firehose) are using this base sink.
+
+# Metrics
+
+There are 3 metrics that implementing sinks will automatically benefit from and therefore shouldn't implement themselves.
+
+* CurrentSendTime Gauge - returns the amount of time in milliseconds it took for the most recent request to write records to return, whether successful or not.
+* NumBytesOut Counter - counts the total number of bytes the sink has tried to write to the destination, using the method `getSizeInBytes` to determine the size of each record. This will double count failures that may need to be retried.
+* NumRecordsOut Counter - similar to above, this counts the total number of records the sink has tried to write to the destination. This will double count failures that may need to be retried.
+
+# Sink Behaviour
+
+
+There are 6 sink configuration settings that control the buffering, flushing and retry behaviour of the sink.
Review comment:
```suggestion
# Sink Behavior
There are six sink configuration settings that control the buffering, flushing, and retry behavior of the sink.
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] CrynetLogistics commented on pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
CrynetLogistics commented on pull request #517:
URL: https://github.com/apache/flink-web/pull/517#issuecomment-1071118511
Attaching a printout
[Apache Flink_ Asynchronous Base Sink.pdf](https://github.com/apache/flink-web/files/8288448/Apache.Flink_.Asynchronous.Base.Sink.pdf)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] CrynetLogistics commented on a change in pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
CrynetLogistics commented on a change in pull request #517:
URL: https://github.com/apache/flink-web/pull/517#discussion_r831017589
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
+
+New for [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the Async Base Sink - an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
+
+This common abstraction will reduce the effort required to maintain all these individual sinks, with bugfixes and improvements to the sink core benefiting all implementations that extend it.
+
+**Attention** The sink is designed to participate in checkpointing to provide at-least once semantics, but it is limited to destinations that provide a client that supports async requests.
+
+The design of the sink focuses on extensibility and a broad support of destinations. The core of the sink is kept generic and free of any connector specific dependencies.
+
+
+{% toc %}
+
+# Dependency
+To use this base sink, add the following dependency to your project:
+
+```xml
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.version}</version>
+</dependency>
+```
+
+
+
+# Public Interfaces
+
+## Generic Types
+
+`<InputT>` – type of elements in a DataStream that should be passed to the sink
+
+`<RequestEntryT>` – type of a payload containing the element and additional metadata that is required to submit a single element to the destination
+
+
+## Element Converter Interface
+```java
+public interface ElementConverter<InputT, RequestEntryT> extends Serializable {
+ RequestEntryT apply(InputT element, SinkWriter.Context context);
+}
+```
+Concrete sink implementers should provide a way to convert from an element in the DataStream to the payload type that contains all the additional metadata required to submit that element to the destination by the sink.
+Ideally this would be hidden from the end user as it allows concrete sink implementers to adapt to changes in the destination api without breaking end user code.
+
+
+## Sink Writer Interface
+
+```java
+public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable>
+ implements SinkWriter<InputT, Void, Collection<RequestEntryT>> {
+ // ...
+ protected abstract void submitRequestEntries
+ (List<RequestEntryT> requestEntries, Consumer<List<RequestEntryT>> requestResult);
+ // ...
+}
+```
+
+This method should specify how a list of elements from the datastream may be persisted into the destination. Sink implementers of various datastore and data processing vendors may use their own clients in connecting to and persisting the requestEntries received by this method.
+
+Should any elements fail to be persisted, they should be requeued back in the buffer for retry using `requestResult.accept(...list of failed entries...)`. However, retrying any element that is known to be faulty and consistently failing, will result in that element being requeued forever, therefore a sensible strategy for determining what should be retried is highly recommended.
+
+```java
+public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable>
+ implements SinkWriter<InputT, Void, Collection<RequestEntryT>> {
+ // ...
+ protected abstract long getSizeInBytes(RequestEntryT requestEntry);
+ // ...
+}
+```
+The generic sink has a concept of size of elements in the buffer. This allows users to specify a byte size threshold beyond which elements will be flushed. However the sink implementer is best positioned to determine what is most sensible measure of size for each `RequestEntryT` is.
Review comment:
Instead of "The generic sink", put "The async sink"
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
+
+New for [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the Async Base Sink - an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
+
+This common abstraction will reduce the effort required to maintain all these individual sinks, with bugfixes and improvements to the sink core benefiting all implementations that extend it.
+
+**Attention** The sink is designed to participate in checkpointing to provide at-least once semantics, but it is limited to destinations that provide a client that supports async requests.
+
+The design of the sink focuses on extensibility and a broad support of destinations. The core of the sink is kept generic and free of any connector specific dependencies.
+
+
+{% toc %}
+
+# Dependency
+To use this base sink, add the following dependency to your project:
+
+```xml
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.version}</version>
+</dependency>
+```
+
+
+
+# Public Interfaces
+
+## Generic Types
+
+`<InputT>` – type of elements in a DataStream that should be passed to the sink
+
+`<RequestEntryT>` – type of a payload containing the element and additional metadata that is required to submit a single element to the destination
+
+
+## Element Converter Interface
+```java
+public interface ElementConverter<InputT, RequestEntryT> extends Serializable {
+ RequestEntryT apply(InputT element, SinkWriter.Context context);
+}
+```
+Concrete sink implementers should provide a way to convert from an element in the DataStream to the payload type that contains all the additional metadata required to submit that element to the destination by the sink.
+Ideally this would be hidden from the end user as it allows concrete sink implementers to adapt to changes in the destination api without breaking end user code.
+
+
+## Sink Writer Interface
+
+```java
+public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable>
+ implements SinkWriter<InputT, Void, Collection<RequestEntryT>> {
+ // ...
+ protected abstract void submitRequestEntries
+ (List<RequestEntryT> requestEntries, Consumer<List<RequestEntryT>> requestResult);
+ // ...
+}
+```
+
+This method should specify how a list of elements from the datastream may be persisted into the destination. Sink implementers of various datastore and data processing vendors may use their own clients in connecting to and persisting the requestEntries received by this method.
+
+Should any elements fail to be persisted, they should be requeued back in the buffer for retry using `requestResult.accept(...list of failed entries...)`. However, retrying any element that is known to be faulty and consistently failing, will result in that element being requeued forever, therefore a sensible strategy for determining what should be retried is highly recommended.
+
+```java
+public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable>
+ implements SinkWriter<InputT, Void, Collection<RequestEntryT>> {
+ // ...
+ protected abstract long getSizeInBytes(RequestEntryT requestEntry);
+ // ...
+}
+```
+The generic sink has a concept of size of elements in the buffer. This allows users to specify a byte size threshold beyond which elements will be flushed. However the sink implementer is best positioned to determine what is most sensible measure of size for each `RequestEntryT` is.
+
+
+```java
+public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable>
+ implements SinkWriter<InputT, Void, Collection<RequestEntryT>> {
+ // ...
+ public AsyncSinkWriter(
+ ElementConverter<InputT, RequestEntryT> elementConverter,
+ Sink.InitContext context,
+ int maxBatchSize,
+ int maxInFlightRequests,
+ int maxBufferedRequests,
+ long flushOnBufferSizeInBytes,
+ long maxTimeInBufferMS,
+ long maxRecordSizeInBytes) { /* ... */ }
+ // ...
+}
+```
+
+## Sink Interface
+
+```java
+public class MySink<InputT> extends AsyncSinkBase<InputT, RequestEntryT> {
+ // ...
+ @Override
+ public SinkWriter<InputT, Void, Collection<PutRecordsRequestEntry>> createWriter
+ (InitContext context, List<Collection<PutRecordsRequestEntry>> states) {
+ return new MySinkWriter(context);
+ }
+ // ...
+}
+```
+Sink implementers extending this will need to return their own extension of the `AsyncSinkWriter` from `createWriter()` inside their own implementation of `AsyncSinkBase`.
+
+Currently the [Kinesis Data Streams sink](https://github.com/apache/flink/tree/master/flink-connectors/flink-connector-aws-kinesis-streams) and [Kinesis Data Firehose sink](https://github.com/apache/flink/tree/master/flink-connectors/flink-connector-aws-kinesis-firehose) are using this base sink.
Review comment:
Use "At the time of writing..." instead of "Currently..."
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] infoverload commented on a change in pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
infoverload commented on a change in pull request #517:
URL: https://github.com/apache/flink-web/pull/517#discussion_r838144107
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
+
+New for [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the Async Base Sink - an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
+
+This common abstraction will reduce the effort required to maintain all these individual sinks, with bugfixes and improvements to the sink core benefiting all implementations that extend it.
+
+**Attention** The sink is designed to participate in checkpointing to provide at-least once semantics, but it is limited to destinations that provide a client that supports async requests.
+
+The design of the sink focuses on extensibility and a broad support of destinations. The core of the sink is kept generic and free of any connector specific dependencies.
+
+
+{% toc %}
+
+# Dependency
+To use this base sink, add the following dependency to your project:
+
Review comment:
```suggestion
# Adding the base sink as a dependency
In order to this base sink, you will need to add the following dependency to your project. The example below follows the Maven syntax:
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] infoverload commented on a change in pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
infoverload commented on a change in pull request #517:
URL: https://github.com/apache/flink-web/pull/517#discussion_r838143218
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
+
+New for [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the Async Base Sink - an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
+
+This common abstraction will reduce the effort required to maintain all these individual sinks, with bugfixes and improvements to the sink core benefiting all implementations that extend it.
+
+**Attention** The sink is designed to participate in checkpointing to provide at-least once semantics, but it is limited to destinations that provide a client that supports async requests.
+
+The design of the sink focuses on extensibility and a broad support of destinations. The core of the sink is kept generic and free of any connector specific dependencies.
Review comment:
```suggestion
The design of the sink focuses on extensibility and a broad support of destinations. The core of the sink is kept generic and free of any connector-specific dependencies.
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] dannycranmer commented on a change in pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
dannycranmer commented on a change in pull request #517:
URL: https://github.com/apache/flink-web/pull/517#discussion_r832979017
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
Review comment:
> They batch records according to user defined buffering hints
I would say this is not a "basic functionalities" of sink, this is a more advanced feature. A basic sink would just send records one by one
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
+
+New for [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the Async Base Sink - an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
Review comment:
> is the Async Base Sink - an ab
nit:
> is the Async Base Sink; an ab
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
Review comment:
General comment, there is some good overview text in the FLIP that you could reuse to give context here
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
+
+New for [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the Async Base Sink - an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
+
+This common abstraction will reduce the effort required to maintain all these individual sinks, with bugfixes and improvements to the sink core benefiting all implementations that extend it.
+
+**Attention** The sink is designed to participate in checkpointing to provide at-least once semantics, but it is limited to destinations that provide a client that supports async requests.
+
+The design of the sink focuses on extensibility and a broad support of destinations. The core of the sink is kept generic and free of any connector specific dependencies.
+
+
+{% toc %}
+
+# Dependency
+To use this base sink, add the following dependency to your project:
+
+```xml
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.version}</version>
+</dependency>
+```
+
+
+
+# Public Interfaces
+
+## Generic Types
+
+`<InputT>` – type of elements in a DataStream that should be passed to the sink
+
+`<RequestEntryT>` – type of a payload containing the element and additional metadata that is required to submit a single element to the destination
+
+
+## Element Converter Interface
+```java
+public interface ElementConverter<InputT, RequestEntryT> extends Serializable {
+ RequestEntryT apply(InputT element, SinkWriter.Context context);
+}
+```
+Concrete sink implementers should provide a way to convert from an element in the DataStream to the payload type that contains all the additional metadata required to submit that element to the destination by the sink.
+Ideally this would be hidden from the end user as it allows concrete sink implementers to adapt to changes in the destination api without breaking end user code.
+
+
+## Sink Writer Interface
+
+```java
+public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable>
+ implements SinkWriter<InputT, Void, Collection<RequestEntryT>> {
+ // ...
+ protected abstract void submitRequestEntries
+ (List<RequestEntryT> requestEntries, Consumer<List<RequestEntryT>> requestResult);
+ // ...
+}
+```
+
+This method should specify how a list of elements from the datastream may be persisted into the destination. Sink implementers of various datastore and data processing vendors may use their own clients in connecting to and persisting the requestEntries received by this method.
Review comment:
> This method should specify how a list of elements from the datastream may be persisted into the destination
I thought this method writes elements to the destination? This is worded in a confusing way, sounds like it should return an Object to tell something else how to write.
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
+
+New for [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the Async Base Sink - an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
+
+This common abstraction will reduce the effort required to maintain all these individual sinks, with bugfixes and improvements to the sink core benefiting all implementations that extend it.
+
+**Attention** The sink is designed to participate in checkpointing to provide at-least once semantics, but it is limited to destinations that provide a client that supports async requests.
Review comment:
> but it is limited to destinations that provide a client that supports async requests.
nit: Technically it is not, you could manage a thread pool in the concrete implementation and use a sync client
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
+
+New for [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the Async Base Sink - an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
+
+This common abstraction will reduce the effort required to maintain all these individual sinks, with bugfixes and improvements to the sink core benefiting all implementations that extend it.
Review comment:
> maintain all these individual sinks
Which individual sinks?
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
+
+New for [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the Async Base Sink - an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
+
+This common abstraction will reduce the effort required to maintain all these individual sinks, with bugfixes and improvements to the sink core benefiting all implementations that extend it.
+
+**Attention** The sink is designed to participate in checkpointing to provide at-least once semantics, but it is limited to destinations that provide a client that supports async requests.
+
+The design of the sink focuses on extensibility and a broad support of destinations. The core of the sink is kept generic and free of any connector specific dependencies.
+
+
+{% toc %}
+
+# Dependency
+To use this base sink, add the following dependency to your project:
+
+```xml
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.version}</version>
+</dependency>
+```
+
+
+
+# Public Interfaces
+
+## Generic Types
+
+`<InputT>` – type of elements in a DataStream that should be passed to the sink
+
+`<RequestEntryT>` – type of a payload containing the element and additional metadata that is required to submit a single element to the destination
+
+
+## Element Converter Interface
+```java
+public interface ElementConverter<InputT, RequestEntryT> extends Serializable {
+ RequestEntryT apply(InputT element, SinkWriter.Context context);
+}
+```
+Concrete sink implementers should provide a way to convert from an element in the DataStream to the payload type that contains all the additional metadata required to submit that element to the destination by the sink.
+Ideally this would be hidden from the end user as it allows concrete sink implementers to adapt to changes in the destination api without breaking end user code.
+
+
+## Sink Writer Interface
+
+```java
+public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable>
+ implements SinkWriter<InputT, Void, Collection<RequestEntryT>> {
Review comment:
This is out of date:
https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriter.java#L54
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
+
+New for [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the Async Base Sink - an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
+
+This common abstraction will reduce the effort required to maintain all these individual sinks, with bugfixes and improvements to the sink core benefiting all implementations that extend it.
+
+**Attention** The sink is designed to participate in checkpointing to provide at-least once semantics, but it is limited to destinations that provide a client that supports async requests.
+
+The design of the sink focuses on extensibility and a broad support of destinations. The core of the sink is kept generic and free of any connector specific dependencies.
+
+
+{% toc %}
+
+# Dependency
+To use this base sink, add the following dependency to your project:
+
+```xml
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.version}</version>
+</dependency>
+```
+
+
+
+# Public Interfaces
+
+## Generic Types
+
+`<InputT>` – type of elements in a DataStream that should be passed to the sink
+
+`<RequestEntryT>` – type of a payload containing the element and additional metadata that is required to submit a single element to the destination
+
+
+## Element Converter Interface
+```java
+public interface ElementConverter<InputT, RequestEntryT> extends Serializable {
+ RequestEntryT apply(InputT element, SinkWriter.Context context);
+}
+```
+Concrete sink implementers should provide a way to convert from an element in the DataStream to the payload type that contains all the additional metadata required to submit that element to the destination by the sink.
+Ideally this would be hidden from the end user as it allows concrete sink implementers to adapt to changes in the destination api without breaking end user code.
+
+
+## Sink Writer Interface
Review comment:
You did not cover how to snapshot state and handle errors
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] infoverload commented on a change in pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
infoverload commented on a change in pull request #517:
URL: https://github.com/apache/flink-web/pull/517#discussion_r838141023
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
Review comment:
```suggestion
excerpt: An overview of the new AsyncBaseSink and how to use it for building your own concrete sink
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] infoverload commented on a change in pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
infoverload commented on a change in pull request #517:
URL: https://github.com/apache/flink-web/pull/517#discussion_r838148551
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
+
+New for [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the Async Base Sink - an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
+
+This common abstraction will reduce the effort required to maintain all these individual sinks, with bugfixes and improvements to the sink core benefiting all implementations that extend it.
+
+**Attention** The sink is designed to participate in checkpointing to provide at-least once semantics, but it is limited to destinations that provide a client that supports async requests.
+
+The design of the sink focuses on extensibility and a broad support of destinations. The core of the sink is kept generic and free of any connector specific dependencies.
+
+
+{% toc %}
+
+# Dependency
+To use this base sink, add the following dependency to your project:
+
+```xml
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.version}</version>
+</dependency>
+```
+
+
+
+# Public Interfaces
+
+## Generic Types
+
+`<InputT>` – type of elements in a DataStream that should be passed to the sink
+
+`<RequestEntryT>` – type of a payload containing the element and additional metadata that is required to submit a single element to the destination
+
+
+## Element Converter Interface
+```java
+public interface ElementConverter<InputT, RequestEntryT> extends Serializable {
+ RequestEntryT apply(InputT element, SinkWriter.Context context);
+}
+```
+Concrete sink implementers should provide a way to convert from an element in the DataStream to the payload type that contains all the additional metadata required to submit that element to the destination by the sink.
+Ideally this would be hidden from the end user as it allows concrete sink implementers to adapt to changes in the destination api without breaking end user code.
+
+
+## Sink Writer Interface
+
+```java
+public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable>
+ implements SinkWriter<InputT, Void, Collection<RequestEntryT>> {
+ // ...
+ protected abstract void submitRequestEntries
+ (List<RequestEntryT> requestEntries, Consumer<List<RequestEntryT>> requestResult);
+ // ...
+}
+```
+
+This method should specify how a list of elements from the datastream may be persisted into the destination. Sink implementers of various datastore and data processing vendors may use their own clients in connecting to and persisting the requestEntries received by this method.
+
+Should any elements fail to be persisted, they should be requeued back in the buffer for retry using `requestResult.accept(...list of failed entries...)`. However, retrying any element that is known to be faulty and consistently failing, will result in that element being requeued forever, therefore a sensible strategy for determining what should be retried is highly recommended.
+
+```java
+public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable>
+ implements SinkWriter<InputT, Void, Collection<RequestEntryT>> {
+ // ...
+ protected abstract long getSizeInBytes(RequestEntryT requestEntry);
+ // ...
+}
+```
+The generic sink has a concept of size of elements in the buffer. This allows users to specify a byte size threshold beyond which elements will be flushed. However the sink implementer is best positioned to determine what is most sensible measure of size for each `RequestEntryT` is.
+
+
+```java
+public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable>
+ implements SinkWriter<InputT, Void, Collection<RequestEntryT>> {
+ // ...
+ public AsyncSinkWriter(
+ ElementConverter<InputT, RequestEntryT> elementConverter,
+ Sink.InitContext context,
+ int maxBatchSize,
+ int maxInFlightRequests,
+ int maxBufferedRequests,
+ long flushOnBufferSizeInBytes,
+ long maxTimeInBufferMS,
+ long maxRecordSizeInBytes) { /* ... */ }
+ // ...
+}
+```
+
+## Sink Interface
+
+```java
+public class MySink<InputT> extends AsyncSinkBase<InputT, RequestEntryT> {
+ // ...
+ @Override
+ public SinkWriter<InputT, Void, Collection<PutRecordsRequestEntry>> createWriter
+ (InitContext context, List<Collection<PutRecordsRequestEntry>> states) {
+ return new MySinkWriter(context);
+ }
+ // ...
+}
+```
+Sink implementers extending this will need to return their own extension of the `AsyncSinkWriter` from `createWriter()` inside their own implementation of `AsyncSinkBase`.
+
+Currently the [Kinesis Data Streams sink](https://github.com/apache/flink/tree/master/flink-connectors/flink-connector-aws-kinesis-streams) and [Kinesis Data Firehose sink](https://github.com/apache/flink/tree/master/flink-connectors/flink-connector-aws-kinesis-firehose) are using this base sink.
+
+# Metrics
+
+There are 3 metrics that implementing sinks will automatically benefit from and therefore shouldn't implement themselves.
Review comment:
```suggestion
There are three metrics that automatically exist when you implement sinks (and, thus, should not be implemented by yourself).
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] infoverload commented on a change in pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
infoverload commented on a change in pull request #517:
URL: https://github.com/apache/flink-web/pull/517#discussion_r838146254
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
+
+New for [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the Async Base Sink - an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
+
+This common abstraction will reduce the effort required to maintain all these individual sinks, with bugfixes and improvements to the sink core benefiting all implementations that extend it.
+
+**Attention** The sink is designed to participate in checkpointing to provide at-least once semantics, but it is limited to destinations that provide a client that supports async requests.
+
+The design of the sink focuses on extensibility and a broad support of destinations. The core of the sink is kept generic and free of any connector specific dependencies.
+
+
+{% toc %}
+
+# Dependency
+To use this base sink, add the following dependency to your project:
+
+```xml
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.version}</version>
+</dependency>
+```
+
+
+
+# Public Interfaces
+
+## Generic Types
+
+`<InputT>` – type of elements in a DataStream that should be passed to the sink
+
+`<RequestEntryT>` – type of a payload containing the element and additional metadata that is required to submit a single element to the destination
+
+
+## Element Converter Interface
+```java
+public interface ElementConverter<InputT, RequestEntryT> extends Serializable {
+ RequestEntryT apply(InputT element, SinkWriter.Context context);
+}
+```
+Concrete sink implementers should provide a way to convert from an element in the DataStream to the payload type that contains all the additional metadata required to submit that element to the destination by the sink.
+Ideally this would be hidden from the end user as it allows concrete sink implementers to adapt to changes in the destination api without breaking end user code.
Review comment:
```suggestion
## ElementConverter Interface
```java
public interface ElementConverter<InputT, RequestEntryT> extends Serializable {
RequestEntryT apply(InputT element, SinkWriter.Context context);
}
```
Classes that implement the concrete sink should provide a way to convert from an element in the DataStream to the payload type that contains all the additional metadata required to submit that element to the destination by the sink. Ideally, this would be hidden from the end user since it allows concrete sink implementers to adapt to changes in the destination API without breaking end user code.
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] CrynetLogistics commented on pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
CrynetLogistics commented on pull request #517:
URL: https://github.com/apache/flink-web/pull/517#issuecomment-1071118511
Attaching a printout
[Apache Flink_ Asynchronous Base Sink.pdf](https://github.com/apache/flink-web/files/8288448/Apache.Flink_.Asynchronous.Base.Sink.pdf)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] infoverload commented on a change in pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
infoverload commented on a change in pull request #517:
URL: https://github.com/apache/flink-web/pull/517#discussion_r838140153
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
Review comment:
```suggestion
date: 2022-04-04 16:00:00
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] infoverload commented on a change in pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
infoverload commented on a change in pull request #517:
URL: https://github.com/apache/flink-web/pull/517#discussion_r838144602
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
+
+New for [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the Async Base Sink - an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
+
+This common abstraction will reduce the effort required to maintain all these individual sinks, with bugfixes and improvements to the sink core benefiting all implementations that extend it.
+
+**Attention** The sink is designed to participate in checkpointing to provide at-least once semantics, but it is limited to destinations that provide a client that supports async requests.
+
+The design of the sink focuses on extensibility and a broad support of destinations. The core of the sink is kept generic and free of any connector specific dependencies.
+
+
+{% toc %}
+
+# Dependency
+To use this base sink, add the following dependency to your project:
+
+```xml
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.version}</version>
+</dependency>
+```
+
+
+
+# Public Interfaces
Review comment:
```suggestion
# The public interfaces of AsyncSinkBase
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-web] infoverload commented on a change in pull request #517: [FLINK-24370] Addition of blog post describing features and usage of the Async Sink…
Posted by GitBox <gi...@apache.org>.
infoverload commented on a change in pull request #517:
URL: https://github.com/apache/flink-web/pull/517#discussion_r838142903
##########
File path: _posts/2022-03-16-async-sink-base.md
##########
@@ -0,0 +1,138 @@
+---
+layout: post
+title: "Asynchronous Base Sink"
+date: 2022-03-17 16:00:00
+authors:
+- CrynetLogistics:
+ name: "Zichen Liu"
+ twitter: "CrynetLogistics"
+excerpt: An overview of the new features of the new Async Base Sink and pointers for building your own concrete sink atop
+---
+
+The basic functionalities of sinks in general are quite similar. They batch records according to user defined buffering hints, sign requests, write them to the destination, retry unsuccessful or throttled requests, and participate in checkpointing.
+
+New for [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the Async Base Sink - an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
Review comment:
```suggestion
New in [Flink 1.15](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) is the AsyncBaseSink, an abstract sink with a number of common functionalities extracted. Adding support for a new destination now only requires a lightweight shim that implements the specific interfaces of the destination using a client that supports async requests.
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org