Merge branch 'master' into tzhang-si-readme

Author: sfc-gh-xhuang

src/main/java/net/snowflake/ingest/streaming/SnowflakeStreamingIngestChannel.java

@@ -9,6 +9,7 @@
 import java.util.concurrent.CompletableFuture;
 import javax.annotation.Nullable;
 import net.snowflake.ingest.streaming.internal.ColumnProperties;
+import net.snowflake.ingest.utils.SFException;
 
 /**
  * A logical partition that represents a connection to a single Snowflake table, data will be
@@ -67,17 +68,27 @@ public interface SnowflakeStreamingIngestChannel {
   String getFullyQualifiedTableName();
 
   /**
-   * @return a boolean which indicates whether the channel is valid
+   * @return a boolean which indicates whether the channel is valid. Typically, this means whether
+   *     the current instance is the owner of the channel, i.e., if another Client opens a {@link
+   *     SnowflakeStreamingIngestChannel} of the same name then the current instance will be
+   *     considered "invalid" as its persisted epoch will have increased. If this returns false then
+   *     calling `insertRow(s)` on this channel instance will result in an {@link SFException} and
+   *     no further writes to the channel will be accepted.
+   *     <p>>Note: there may be a delay between server-side invalidations and the Client detecting
+   *     it so this may not immediately return false in the event of a server-side invalidation.
    */
   boolean isValid();
 
   /**
-   * @return a boolean which indicates whether the channel is closed
+   * @return a boolean which indicates whether the channel is closed. If true this means that the
+   *     current {@link SnowflakeStreamingIngestChannel} will not accept any additional rows on
+   *     calls to `insertRow(s)`
    */
   boolean isClosed();
 
   /**
-   * Close the channel, this function will make sure all the data in this channel is committed
+   * Close the channel. Closing entails draining any outstanding data in the Channel's buffer and
+   * marking the Channel as no longer being able to accept writes via `insertRow(s)`*
    *
    * @return a completable future which will be completed when the channel is closed
    */
@@ -206,7 +217,7 @@ public interface SnowflakeStreamingIngestChannel {
    *                 </li>
    *
    *             </ul>
-   *
+   * <p>
    *             For TIMESTAMP_LTZ and TIMESTAMP_TZ, all input without timezone will be by default interpreted in the timezone "America/Los_Angeles". This can be changed by calling {@link net.snowflake.ingest.streaming.OpenChannelRequest.OpenChannelRequestBuilder#setDefaultTimezone(ZoneId)}.
    *         </td>
    *         <tr>
@@ -248,6 +259,9 @@ public interface SnowflakeStreamingIngestChannel {
    * @param offsetToken offset of given row, used for replay in case of failures. It could be null
    *     if you don't plan on replaying or can't replay
    * @return insert response that possibly contains errors because of insertion failures
+   * @throws SFException if the channel is not considered "valid". Typically, this means that
+   *     another Client has claimed ownership of the Channel. Writes to this channel will be
+   *     rejected and result in this exception being thrown.
    */
   InsertValidationResponse insertRow(Map<String, Object> row, @Nullable String offsetToken);
 
@@ -262,6 +276,9 @@ public interface SnowflakeStreamingIngestChannel {
    * @param endOffsetToken end offset of the batch/row-set, used for replay in case of failures, *
    *     It could be null if you don't plan on replaying or can't replay
    * @return insert response that possibly contains errors because of insertion failures
+   * @throws SFException if the channel is not considered "valid". Typically, this means that
+   *     another Client has claimed ownership of the Channel. Writes to this channel will be
+   *     rejected and result in this exception being thrown.
    */
   InsertValidationResponse insertRows(
       Iterable<Map<String, Object>> rows,
@@ -271,6 +288,10 @@ InsertValidationResponse insertRows(
   /**
    * Insert a batch of rows into the channel with the end offset token only, please see {@link
    * SnowflakeStreamingIngestChannel#insertRows(Iterable, String, String)} for more information.
+   *
+   * @throws SFException if the channel is not considered "valid". Typically, this means that
+   *     another Client has claimed ownership of the Channel. Writes to this channel will be
+   *     rejected and result in this exception being thrown.
    */
   InsertValidationResponse insertRows(
       Iterable<Map<String, Object>> rows, @Nullable String offsetToken);
@@ -279,6 +300,11 @@ InsertValidationResponse insertRows(
    * Get the latest committed offset token from Snowflake
    *
    * @return the latest committed offset token
+   * @throws SFException if Snowflake returns an invalid response code from its `status` endpoint
+   *     which typically indicates that the channel needs to be re-opened. This is evaluated with
+   *     respect to the current Client epoch, i.e. if another Client opens the channel with the same
+   *     name this will throw an exception as the current instant is not the active version of the
+   *     channel.
    */
   @Nullable
   String getLatestCommittedOffsetToken();

src/main/java/net/snowflake/ingest/streaming/internal/ChannelCache.java

@@ -43,6 +43,12 @@ static class FlushInfo {
   /**
    * Add a channel to the channel cache
    *
+   * <p>Note: if there was a previous instance of the channel then the old one is considered
+   * "invalid". Callers with a reference to the old channel object will have their writes rejected
+   * as the channel reference will be marked "invalid". Similarly, calls to fetch the current status
+   * of old versions of the channel will have an exception thrown as the channel is considered
+   * invalid.
+   *
    * @param channel
    */
   void addChannel(SnowflakeStreamingIngestChannelInternal<T> channel) {

src/main/java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestChannelInternal.java

@@ -220,7 +220,7 @@ public boolean isClosed() {
     return this.isClosed;
   }
 
-  /** Mark the channel as closed */
+  /** Mark the channel as closed, meaning that no additional writes can be made to the channel */
   void markClosed() {
     this.isClosed = true;
     logger.logInfo(
@@ -262,6 +262,14 @@ public CompletableFuture<Void> close() {
     return this.close(false);
   }
 
+  /**
+   * Attempts to close the channel by draining any outstanding data and rejecting further writes to
+   * the channel.
+   *
+   * @param drop if true, the channel will be dropped after all data is successfully committed.
+   * @throws SFException when the channel is invalid or closed, meaning that no new rows can be
+   *     inserted via this instance of the channel.
+   */
   @Override
   public CompletableFuture<Void> close(boolean drop) {
     checkValidation();
@@ -302,7 +310,7 @@ public CompletableFuture<Void> close(boolean drop) {
   }
 
   /**
-   * Setup the column fields and vectors using the column metadata from the server
+   * Set up the column fields and vectors using the column metadata from the server
    *
    * @param columns
    */
@@ -322,10 +330,15 @@ void setupSchema(List<ColumnMetadata> columns) {
   /**
    * The row is represented using Map where the key is column name and the value is a row of data
    *
+   * <p>Notes: this may throttle on the calling thread if there is insufficient memory to accept the
+   * supplied rows. This is done by sleeping in the current thread and checking memory conditions.
+   * Thus, the caller should expect this to not return immediately.
+   *
    * @param row object data to write
    * @param offsetToken offset of given row, used for replay in case of failures
    * @return insert response that possibly contains errors because of insertion failures
-   * @throws SFException when the channel is invalid or closed
+   * @throws SFException when the channel is invalid or closed, meaning that no new rows can be
+   *     inserted via this instance of the channel.
    */
   @Override
   public InsertValidationResponse insertRow(Map<String, Object> row, String offsetToken) {
@@ -344,11 +357,19 @@ public InsertValidationResponse insertRow(Map<String, Object> row, String offset
    * SnowflakeStreamingIngestChannel#insertRow(Map, String)} for more information about accepted
    * values.
    *
+   * <p>Notes: this may throttle on the calling thread if there is insufficient memory to accept the
+   * supplied rows. This is done by sleeping in the current thread and checking memory conditions.
+   * Thus, the caller should expect this to not return immediately.
+   *
    * @param rows object data to write
    * @param startOffsetToken start offset of the batch/row-set
    * @param endOffsetToken end offset of the batch/row-set, used for replay in case of failures, *
    *     It could be null if you don't plan on replaying or can't replay
-   * @return insert response that possibly contains errors because of insertion failures
+   * @return insert response that possibly contains errors because of insertion failures throws
+   *     SFException if the channel is closed or is considered "invalid", i.e. another Client
+   *     instance has opened the Channel.
+   * @throws SFException when the channel is invalid or closed, meaning that no new rows can be
+   *     inserted via this instance of the channel.
    */
   @Override
   public InsertValidationResponse insertRows(
@@ -401,11 +422,12 @@ void collectRowSize(float rowSize) {
   }
 
   /**
-   * Get the latest committed offset token from Snowflake, an exception will be thrown if the
-   * channel becomes invalid due to errors and the channel needs to be reopened in order to return a
-   * valid offset token
+   * Get the latest committed offset token from Snowflake for this Channel instance.
    *
-   * @return the latest committed offset token
+   * @return the latest committed offset token for this Channel instance.
+   * @throws SFException if the current instance of the Channel is not the current owner OR if the
+   *     caller cannot communicate with Snowflake. Typically, this indicates that another Client
+   *     instance has opened up the Channel with the same name.
    */
   @Override
   public String getLatestCommittedOffsetToken() {
@@ -427,7 +449,11 @@ public Map<String, ColumnProperties> getTableSchema() {
     return this.tableColumns;
   }
 
-  /** Check whether we need to throttle the insertRows API */
+  /**
+   * Potentially throttles a call to the `insertRow(s)` APIs if needed. Note that throttling applies
+   * to the calling thread by sleeping on the curernt thread and checking whether memory conditions
+   * are suitable to accepting the current row.
+   */
   void throttleInsertIfNeeded(MemoryInfoProvider memoryInfoProvider) {
     int retry = 0;
     while ((hasLowRuntimeMemory(memoryInfoProvider)
@@ -471,7 +497,12 @@ private boolean hasLowRuntimeMemory(MemoryInfoProvider memoryInfoProvider) {
     return hasLowRuntimeMemory;
   }
 
-  /** Check whether the channel is still valid, cleanup and throw an error if not */
+  /**
+   * Check whether the channel is considered "valid", i.e. the current instance of the channel
+   * matches the server-side persisted Client epoch.
+   *
+   * @throws SFException if the current instance of the channel is no longer considered valid
+   */
   private void checkValidation() {
     if (!isValid()) {
       this.owningClient.removeChannelIfSequencersMatch(this);

src/main/java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestClientInternal.java

@@ -481,10 +481,11 @@ public Map<String, String> getLatestCommittedOffsetTokens(
   }
 
   /**
-   * Fetch channels status from Snowflake
+   * Fetch the status of one or more Channels from Snowflake
    *
-   * @param channels a list of channels that we want to get the status on
+   * @param channels a list of channels that we want to get the status of
    * @return a ChannelsStatusResponse object
+   * @throws SFException if the caller cannot communicate with Snowflake
    */
   ChannelsStatusResponse getChannelsStatus(
       List<SnowflakeStreamingIngestChannelInternal<?>> channels) {