001/* 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.client; 019 020import java.io.Closeable; 021import java.util.Collections; 022import java.util.List; 023import java.util.Map; 024import java.util.concurrent.CompletableFuture; 025import java.util.concurrent.TimeUnit; 026import org.apache.hadoop.conf.Configuration; 027import org.apache.hadoop.hbase.TableName; 028import org.apache.yetus.audience.InterfaceAudience; 029 030import org.apache.hbase.thirdparty.com.google.common.collect.Iterables; 031 032/** 033 * Used to communicate with a single HBase table in batches. Obtain an instance from a 034 * {@link AsyncConnection} and call {@link #close()} afterwards. 035 * <p> 036 * The implementation is required to be thread safe. 037 */ 038@InterfaceAudience.Public 039public interface AsyncBufferedMutator extends Closeable { 040 041 /** 042 * Gets the fully qualified table name instance of the table that this 043 * {@code AsyncBufferedMutator} writes to. 044 */ 045 TableName getName(); 046 047 /** 048 * Returns the {@link org.apache.hadoop.conf.Configuration} object used by this instance. 049 * <p> 050 * The reference returned is not a copy, so any change made to it will affect this instance. 051 */ 052 Configuration getConfiguration(); 053 054 /** 055 * Sends a {@link Mutation} to the table. The mutations will be buffered and sent over the wire as 056 * part of a batch. Currently only supports {@link Put} and {@link Delete} mutations. 057 * @param mutation The data to send. 058 */ 059 default CompletableFuture<Void> mutate(Mutation mutation) { 060 return Iterables.getOnlyElement(mutate(Collections.singletonList(mutation))); 061 } 062 063 /** 064 * Send some {@link Mutation}s to the table. The mutations will be buffered and sent over the wire 065 * as part of a batch. There is no guarantee of sending entire content of {@code mutations} in a 066 * single batch, the implementations are free to break it up according to the write buffer 067 * capacity. 068 * @param mutations The data to send. 069 */ 070 List<CompletableFuture<Void>> mutate(List<? extends Mutation> mutations); 071 072 /** 073 * Executes all the buffered, asynchronous operations. 074 */ 075 void flush(); 076 077 /** 078 * Performs a {@link #flush()} and releases any resources held. 079 */ 080 @Override 081 void close(); 082 083 /** 084 * Returns the maximum size in bytes of the write buffer. 085 * <p> 086 * The default value comes from the configuration parameter {@code hbase.client.write.buffer}. 087 * @return The size of the write buffer in bytes. 088 */ 089 long getWriteBufferSize(); 090 091 /** 092 * Returns the periodical flush interval, 0 means disabled. 093 */ 094 default long getPeriodicalFlushTimeout(TimeUnit unit) { 095 throw new UnsupportedOperationException("Not implemented"); 096 } 097 098 /** 099 * Returns the rpc request attributes. 100 */ 101 default Map<String, byte[]> getRequestAttributes() { 102 return Collections.emptyMap(); 103 } 104}