Skip to content

HBASE-27657: Connection and Request Attributes #5326

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bbeaudreault merged 12 commits into from
Jul 24, 2023
Merged

HBASE-27657: Connection and Request Attributes #5326

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
a2d67c3
Connection and Request Attributes
rmdmattingly Mar 2, 2023
441ae90
PR feedback: prefer emptyMap, no wildcard imports
Jun 26, 2023
33481e1
default attributes in Call constructor
Jun 27, 2023
beef1fb
fix RpcClient in TestRpcBasedRegistryHedgedReads
Jun 27, 2023
c3dde33
more test fixes
Jun 27, 2023
d5e092d
request attributes support in tablebuilder
Jul 14, 2023
ee3eef8
cleanup
Jul 17, 2023
0b23de4
checkstyle, banned imports
Jul 18, 2023
99169a2
PR feedback
Jul 18, 2023
2be8a9f
support setting a single req attribute
Jul 19, 2023
e68afb6
remove plural attribute setter
Jul 20, 2023
fe17ca3
fix tests
Jul 20, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion ...-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdminRequestRetryingCaller.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import static org.apache.hadoop.hbase.util.FutureUtils.addListener;

import java.io.IOException;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.ipc.HBaseRpcController;
Expand All @@ -44,7 +45,7 @@ public AsyncAdminRequestRetryingCaller(Timer retryTimer, AsyncConnectionImpl con
long pauseNs, long pauseNsForServerOverloaded, int maxAttempts, long operationTimeoutNs,
long rpcTimeoutNs, int startLogErrorsCnt, ServerName serverName, Callable<T> callable) {
super(retryTimer, conn, priority, pauseNs, pauseNsForServerOverloaded, maxAttempts,
operationTimeoutNs, rpcTimeoutNs, startLogErrorsCnt);
operationTimeoutNs, rpcTimeoutNs, startLogErrorsCnt, Collections.emptyMap());
this.serverName = serverName;
this.callable = callable;
}
Expand Down
7 changes: 6 additions & 1 deletion hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncBatchRpcRetryingCaller.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,8 @@ class AsyncBatchRpcRetryingCaller<T> {

private final HBaseServerExceptionPauseManager pauseManager;

private final Map<String, byte[]> requestAttributes;

// we can not use HRegionLocation as the map key because the hashCode and equals method of
// HRegionLocation only consider serverName.
private static final class RegionRequest {
Expand Down Expand Up @@ -149,7 +151,8 @@ public int getPriority() {

public AsyncBatchRpcRetryingCaller(Timer retryTimer, AsyncConnectionImpl conn,
TableName tableName, List<? extends Row> actions, long pauseNs, long pauseNsForServerOverloaded,
int maxAttempts, long operationTimeoutNs, long rpcTimeoutNs, int startLogErrorsCnt) {
int maxAttempts, long operationTimeoutNs, long rpcTimeoutNs, int startLogErrorsCnt,
Map<String, byte[]> requestAttributes) {
this.retryTimer = retryTimer;
this.conn = conn;
this.tableName = tableName;
Expand Down Expand Up @@ -180,6 +183,7 @@ public AsyncBatchRpcRetryingCaller(Timer retryTimer, AsyncConnectionImpl conn,
this.startNs = System.nanoTime();
this.pauseManager =
new HBaseServerExceptionPauseManager(pauseNs, pauseNsForServerOverloaded, operationTimeoutNs);
this.requestAttributes = requestAttributes;
}

private static boolean hasIncrementOrAppend(Row action) {
Expand Down Expand Up @@ -392,6 +396,7 @@ private void sendToServer(ServerName serverName, ServerRequest serverReq, int tr
HBaseRpcController controller = conn.rpcControllerFactory.newController();
resetController(controller, Math.min(rpcTimeoutNs, remainingNs),
calcPriority(serverReq.getPriority(), tableName));
controller.setRequestAttributes(requestAttributes);
if (!cells.isEmpty()) {
controller.setCellScanner(createCellScanner(cells));
}
Expand Down
31 changes: 19 additions & 12 deletions hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncClientScanner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.context.Scope;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
Expand Down Expand Up @@ -92,9 +93,12 @@ class AsyncClientScanner {

private final Span span;

private final Map<String, byte[]> requestAttributes;

public AsyncClientScanner(Scan scan, AdvancedScanResultConsumer consumer, TableName tableName,
AsyncConnectionImpl conn, Timer retryTimer, long pauseNs, long pauseNsForServerOverloaded,
int maxAttempts, long scanTimeoutNs, long rpcTimeoutNs, int startLogErrorsCnt) {
int maxAttempts, long scanTimeoutNs, long rpcTimeoutNs, int startLogErrorsCnt,
Map<String, byte[]> requestAttributes) {
if (scan.getStartRow() == null) {
scan.withStartRow(EMPTY_START_ROW, scan.includeStartRow());
}
Expand All @@ -113,6 +117,7 @@ public AsyncClientScanner(Scan scan, AdvancedScanResultConsumer consumer, TableN
this.rpcTimeoutNs = rpcTimeoutNs;
this.startLogErrorsCnt = startLogErrorsCnt;
this.resultCache = createScanResultCache(scan);
this.requestAttributes = requestAttributes;
if (scan.isScanMetricsEnabled()) {
this.scanMetrics = new ScanMetrics();
consumer.onScanMetricsCreated(scanMetrics);
Expand Down Expand Up @@ -191,15 +196,17 @@ private CompletableFuture<OpenScannerResponse> callOpenScanner(HBaseRpcControlle
}

private void startScan(OpenScannerResponse resp) {
addListener(conn.callerFactory.scanSingleRegion().id(resp.resp.getScannerId())
.location(resp.loc).remote(resp.isRegionServerRemote)
.scannerLeaseTimeoutPeriod(resp.resp.getTtl(), TimeUnit.MILLISECONDS).stub(resp.stub)
.setScan(scan).metrics(scanMetrics).consumer(consumer).resultCache(resultCache)
.rpcTimeout(rpcTimeoutNs, TimeUnit.NANOSECONDS)
.scanTimeout(scanTimeoutNs, TimeUnit.NANOSECONDS).pause(pauseNs, TimeUnit.NANOSECONDS)
.pauseForServerOverloaded(pauseNsForServerOverloaded, TimeUnit.NANOSECONDS)
.maxAttempts(maxAttempts).startLogErrorsCnt(startLogErrorsCnt)
.start(resp.controller, resp.resp), (hasMore, error) -> {
addListener(
conn.callerFactory.scanSingleRegion().id(resp.resp.getScannerId()).location(resp.loc)
.remote(resp.isRegionServerRemote)
.scannerLeaseTimeoutPeriod(resp.resp.getTtl(), TimeUnit.MILLISECONDS).stub(resp.stub)
.setScan(scan).metrics(scanMetrics).consumer(consumer).resultCache(resultCache)
.rpcTimeout(rpcTimeoutNs, TimeUnit.NANOSECONDS)
.scanTimeout(scanTimeoutNs, TimeUnit.NANOSECONDS).pause(pauseNs, TimeUnit.NANOSECONDS)
.pauseForServerOverloaded(pauseNsForServerOverloaded, TimeUnit.NANOSECONDS)
.maxAttempts(maxAttempts).startLogErrorsCnt(startLogErrorsCnt)
.setRequestAttributes(requestAttributes).start(resp.controller, resp.resp),
(hasMore, error) -> {
try (Scope ignored = span.makeCurrent()) {
if (error != null) {
try {
Expand Down Expand Up @@ -231,8 +238,8 @@ private CompletableFuture<OpenScannerResponse> openScanner(int replicaId) {
.priority(scan.getPriority()).rpcTimeout(rpcTimeoutNs, TimeUnit.NANOSECONDS)
.operationTimeout(scanTimeoutNs, TimeUnit.NANOSECONDS).pause(pauseNs, TimeUnit.NANOSECONDS)
.pauseForServerOverloaded(pauseNsForServerOverloaded, TimeUnit.NANOSECONDS)
.maxAttempts(maxAttempts).startLogErrorsCnt(startLogErrorsCnt).action(this::callOpenScanner)
.call();
.maxAttempts(maxAttempts).startLogErrorsCnt(startLogErrorsCnt)
.setRequestAttributes(requestAttributes).action(this::callOpenScanner).call();
}
}

Expand Down
11 changes: 9 additions & 2 deletions hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnectionImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,8 @@
import io.opentelemetry.api.trace.Span;
import java.io.IOException;
import java.net.SocketAddress;
import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
Expand Down Expand Up @@ -127,6 +129,11 @@ public class AsyncConnectionImpl implements AsyncConnection {

public AsyncConnectionImpl(Configuration conf, ConnectionRegistry registry, String clusterId,
SocketAddress localAddress, User user) {
this(conf, registry, clusterId, localAddress, user, Collections.emptyMap());
}

public AsyncConnectionImpl(Configuration conf, ConnectionRegistry registry, String clusterId,
SocketAddress localAddress, User user, Map<String, byte[]> connectionAttributes) {
this.conf = conf;
this.user = user;
this.metricsScope = MetricsConnection.getScope(conf, clusterId, this);
Expand All @@ -142,8 +149,8 @@ public AsyncConnectionImpl(Configuration conf, ConnectionRegistry registry, Stri
} else {
this.metrics = Optional.empty();
}
this.rpcClient =
RpcClientFactory.createClient(conf, clusterId, localAddress, metrics.orElse(null));
this.rpcClient = RpcClientFactory.createClient(conf, clusterId, localAddress,
metrics.orElse(null), connectionAttributes);
this.rpcControllerFactory = RpcControllerFactory.instantiate(conf);
this.rpcTimeout =
(int) Math.min(Integer.MAX_VALUE, TimeUnit.NANOSECONDS.toMillis(connConf.getRpcTimeoutNs()));
Expand Down
3 changes: 2 additions & 1 deletion ...ent/src/main/java/org/apache/hadoop/hbase/client/AsyncMasterRequestRpcRetryingCaller.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

import static org.apache.hadoop.hbase.util.FutureUtils.addListener;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import org.apache.hadoop.hbase.exceptions.ClientExceptionsUtil;
import org.apache.hadoop.hbase.ipc.HBaseRpcController;
Expand Down Expand Up @@ -47,7 +48,7 @@ public AsyncMasterRequestRpcRetryingCaller(Timer retryTimer, AsyncConnectionImpl
Callable<T> callable, int priority, long pauseNs, long pauseNsForServerOverloaded,
int maxRetries, long operationTimeoutNs, long rpcTimeoutNs, int startLogErrorsCnt) {
super(retryTimer, conn, priority, pauseNs, pauseNsForServerOverloaded, maxRetries,
operationTimeoutNs, rpcTimeoutNs, startLogErrorsCnt);
operationTimeoutNs, rpcTimeoutNs, startLogErrorsCnt, Collections.emptyMap());
this.callable = callable;
}

Expand Down
4 changes: 3 additions & 1 deletion hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncRpcRetryingCaller.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.concurrent.CompletableFuture;
Expand Down Expand Up @@ -78,7 +79,7 @@ public abstract class AsyncRpcRetryingCaller<T> {

public AsyncRpcRetryingCaller(Timer retryTimer, AsyncConnectionImpl conn, int priority,
long pauseNs, long pauseNsForServerOverloaded, int maxAttempts, long operationTimeoutNs,
long rpcTimeoutNs, int startLogErrorsCnt) {
long rpcTimeoutNs, int startLogErrorsCnt, Map<String, byte[]> requestAttributes) {
this.retryTimer = retryTimer;
this.conn = conn;
this.priority = priority;
Expand All @@ -89,6 +90,7 @@ public AsyncRpcRetryingCaller(Timer retryTimer, AsyncConnectionImpl conn, int pr
this.future = new CompletableFuture<>();
this.controller = conn.rpcControllerFactory.newController();
this.controller.setPriority(priority);
this.controller.setRequestAttributes(requestAttributes);
this.exceptions = new ArrayList<>();
this.startNs = System.nanoTime();
this.pauseManager =
Expand Down
31 changes: 28 additions & 3 deletions hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncRpcRetryingCallerFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,9 @@
import static org.apache.hbase.thirdparty.com.google.common.base.Preconditions.checkArgument;
import static org.apache.hbase.thirdparty.com.google.common.base.Preconditions.checkNotNull;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import org.apache.hadoop.hbase.HRegionLocation;
Expand Down Expand Up @@ -83,6 +85,8 @@ public class SingleRequestCallerBuilder<T> extends BuilderBase {

private int priority = PRIORITY_UNSET;

private Map<String, byte[]> requestAttributes = Collections.emptyMap();

public SingleRequestCallerBuilder<T> table(TableName tableName) {
this.tableName = tableName;
return this;
Expand Down Expand Up @@ -144,6 +148,12 @@ public SingleRequestCallerBuilder<T> priority(int priority) {
return this;
}

public SingleRequestCallerBuilder<T>
setRequestAttributes(Map<String, byte[]> requestAttributes) {
this.requestAttributes = requestAttributes;
return this;
}

private void preCheck() {
checkArgument(replicaId >= 0, "invalid replica id %s", replicaId);
checkNotNull(tableName, "tableName is null");
Expand All @@ -157,7 +167,7 @@ public AsyncSingleRequestRpcRetryingCaller<T> build() {
preCheck();
return new AsyncSingleRequestRpcRetryingCaller<>(retryTimer, conn, tableName, row, replicaId,
locateType, callable, priority, pauseNs, pauseNsForServerOverloaded, maxAttempts,
operationTimeoutNs, rpcTimeoutNs, startLogErrorsCnt);
operationTimeoutNs, rpcTimeoutNs, startLogErrorsCnt, requestAttributes);
}

/**
Expand Down Expand Up @@ -201,6 +211,8 @@ public class ScanSingleRegionCallerBuilder extends BuilderBase {

private int priority = PRIORITY_UNSET;

private Map<String, byte[]> requestAttributes = Collections.emptyMap();

public ScanSingleRegionCallerBuilder id(long scannerId) {
this.scannerId = scannerId;
return this;
Expand Down Expand Up @@ -278,6 +290,12 @@ public ScanSingleRegionCallerBuilder startLogErrorsCnt(int startLogErrorsCnt) {
return this;
}

public ScanSingleRegionCallerBuilder
setRequestAttributes(Map<String, byte[]> requestAttributes) {
this.requestAttributes = requestAttributes;
return this;
}

private void preCheck() {
checkArgument(scannerId != null, "invalid scannerId %d", scannerId);
checkNotNull(scan, "scan is null");
Expand All @@ -293,7 +311,7 @@ public AsyncScanSingleRegionRpcRetryingCaller build() {
return new AsyncScanSingleRegionRpcRetryingCaller(retryTimer, conn, scan, scanMetrics,
scannerId, resultCache, consumer, stub, loc, isRegionServerRemote, priority,
scannerLeaseTimeoutPeriodNs, pauseNs, pauseNsForServerOverloaded, maxAttempts,
scanTimeoutNs, rpcTimeoutNs, startLogErrorsCnt);
scanTimeoutNs, rpcTimeoutNs, startLogErrorsCnt, requestAttributes);
}

/**
Expand Down Expand Up @@ -322,6 +340,8 @@ public class BatchCallerBuilder extends BuilderBase {

private long rpcTimeoutNs = -1L;

private Map<String, byte[]> requestAttributes = Collections.emptyMap();

public BatchCallerBuilder table(TableName tableName) {
this.tableName = tableName;
return this;
Expand Down Expand Up @@ -362,10 +382,15 @@ public BatchCallerBuilder startLogErrorsCnt(int startLogErrorsCnt) {
return this;
}

public BatchCallerBuilder setRequestAttributes(Map<String, byte[]> requestAttributes) {
this.requestAttributes = requestAttributes;
return this;
}

public <T> AsyncBatchRpcRetryingCaller<T> build() {
return new AsyncBatchRpcRetryingCaller<>(retryTimer, conn, tableName, actions, pauseNs,
pauseNsForServerOverloaded, maxAttempts, operationTimeoutNs, rpcTimeoutNs,
startLogErrorsCnt);
startLogErrorsCnt, requestAttributes);
}

public <T> List<CompletableFuture<T>> call() {
Expand Down
4 changes: 3 additions & 1 deletion .../src/main/java/org/apache/hadoop/hbase/client/AsyncScanSingleRegionRpcRetryingCaller.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.concurrent.CompletableFuture;
Expand Down Expand Up @@ -316,7 +317,7 @@ public AsyncScanSingleRegionRpcRetryingCaller(Timer retryTimer, AsyncConnectionI
AdvancedScanResultConsumer consumer, Interface stub, HRegionLocation loc,
boolean isRegionServerRemote, int priority, long scannerLeaseTimeoutPeriodNs, long pauseNs,
long pauseNsForServerOverloaded, int maxAttempts, long scanTimeoutNs, long rpcTimeoutNs,
int startLogErrorsCnt) {
int startLogErrorsCnt, Map<String, byte[]> requestAttributes) {
this.retryTimer = retryTimer;
this.conn = conn;
this.scan = scan;
Expand All @@ -341,6 +342,7 @@ public AsyncScanSingleRegionRpcRetryingCaller(Timer retryTimer, AsyncConnectionI
this.priority = priority;
this.controller = conn.rpcControllerFactory.newController();
this.controller.setPriority(priority);
this.controller.setRequestAttributes(requestAttributes);
this.exceptions = new ArrayList<>();
this.pauseManager =
new HBaseServerExceptionPauseManager(pauseNs, pauseNsForServerOverloaded, scanTimeoutNs);
Expand Down
3 changes: 2 additions & 1 deletion ...ent/src/main/java/org/apache/hadoop/hbase/client/AsyncServerRequestRpcRetryingCaller.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import static org.apache.hadoop.hbase.util.FutureUtils.addListener;

import java.io.IOException;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.ServerName;
Expand Down Expand Up @@ -49,7 +50,7 @@ public AsyncServerRequestRpcRetryingCaller(Timer retryTimer, AsyncConnectionImpl
long pauseNs, long pauseNsForServerOverloaded, int maxAttempts, long operationTimeoutNs,
long rpcTimeoutNs, int startLogErrorsCnt, ServerName serverName, Callable<T> callable) {
super(retryTimer, conn, HConstants.NORMAL_QOS, pauseNs, pauseNsForServerOverloaded, maxAttempts,
operationTimeoutNs, rpcTimeoutNs, startLogErrorsCnt);
operationTimeoutNs, rpcTimeoutNs, startLogErrorsCnt, Collections.emptyMap());
this.serverName = serverName;
this.callable = callable;
}
Expand Down
6 changes: 4 additions & 2 deletions ...ent/src/main/java/org/apache/hadoop/hbase/client/AsyncSingleRequestRpcRetryingCaller.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import static org.apache.hadoop.hbase.util.FutureUtils.addListener;

import java.io.IOException;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import org.apache.hadoop.hbase.HRegionLocation;
Expand Down Expand Up @@ -57,9 +58,10 @@ CompletableFuture<T> call(HBaseRpcController controller, HRegionLocation loc,
public AsyncSingleRequestRpcRetryingCaller(Timer retryTimer, AsyncConnectionImpl conn,
TableName tableName, byte[] row, int replicaId, RegionLocateType locateType,
Callable<T> callable, int priority, long pauseNs, long pauseNsForServerOverloaded,
int maxAttempts, long operationTimeoutNs, long rpcTimeoutNs, int startLogErrorsCnt) {
int maxAttempts, long operationTimeoutNs, long rpcTimeoutNs, int startLogErrorsCnt,
Map<String, byte[]> requestAttributes) {
super(retryTimer, conn, priority, pauseNs, pauseNsForServerOverloaded, maxAttempts,
operationTimeoutNs, rpcTimeoutNs, startLogErrorsCnt);
operationTimeoutNs, rpcTimeoutNs, startLogErrorsCnt, requestAttributes);
this.tableName = tableName;
this.row = row;
this.replicaId = replicaId;
Expand Down
7 changes: 7 additions & 0 deletions hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import static org.apache.hadoop.hbase.util.FutureUtils.allOf;

import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
Expand Down Expand Up @@ -110,6 +111,12 @@ public interface AsyncTable<C extends ScanResultConsumerBase> {
*/
long getScanTimeout(TimeUnit unit);

/**
* Get the map of request attributes
* @return a map of request attributes supplied by the client
*/
Map<String, byte[]> getRequestAttributes();
Comment thread
rmdmattingly marked this conversation as resolved.
Outdated

/**
* Test for the existence of columns in the table, as specified by the Get.
* <p>
Expand Down
6 changes: 6 additions & 0 deletions hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTableBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

import static org.apache.hadoop.hbase.client.ConnectionUtils.retries2Attempts;

import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.apache.hadoop.hbase.HBaseServerException;
import org.apache.yetus.audience.InterfaceAudience;
Expand Down Expand Up @@ -137,6 +138,11 @@ default AsyncTableBuilder<C> setMaxRetries(int maxRetries) {
*/
AsyncTableBuilder<C> setStartLogErrorsCnt(int startLogErrorsCnt);

/**
* Sets the map of request attributes
*/
AsyncTableBuilder<C> setRequestAttributes(Map<String, byte[]> requestAttributes);

@bbeaudreault bbeaudreault Jul 19, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What do you think about setRequestAttribute(String, byte[]) so that one can more fluently create their Table/AsyncTable if desired? One can chain together ImmutableMap.builder().add().add().build, but not everyone might have it and it's a bit more verbose.

@bbeaudreault bbeaudreault Jul 19, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We prob could leave all the *Caller classes as you have them since they are internal and post-build, but for the TableBuilder/AsyncTableBuilder it might be nice.

@rmdmattingly rmdmattingly Jul 19, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah I'd be happy to add that. Just to clarify, we should probably support both, right? And calling setRA after addRA would implicitly overwrite the individual additions?

@bbeaudreault bbeaudreault Jul 19, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't have a strong opinion on whether we should continue to support setRequestAttributes (with an s). Do you think that has a particular value as is?

If we do support both, then yea we need to figure out the interplay. I agree that calling set after add would overwrite, but what about add after set? One could imagine someone calling set with an immutable map, so we'd probably have to copy it in order to ensure add operations work.

@rmdmattingly rmdmattingly Jul 19, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One could imagine someone calling set with an immutable map, so we'd probably have to copy it in order to ensure add operations work.

Yeah that's a good point.

I don't have a strong opinion on whether we should continue to support setRequestAttributes (with an s). Do you think that has a particular value as is?

I don't feel particularly strongly, but it seems plausible that someone will have a map of requests pre-constructed, particularly if they have several attributes to pass, and that it would be nicer to pass it in than need to iterate calls to addRA

@rmdmattingly rmdmattingly Jul 19, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we provide a setRequestAttribute(String, byte[]) method then do we also want to support unsetting a key?

@rmdmattingly rmdmattingly Jul 19, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just pushed a change which allows for the two methods setRequestAttribute and setRequestAttributes. The plural will implicitly overwrite anything already submitted. It also enforces the implementation of map that we're working with

@bbeaudreault bbeaudreault Jul 20, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it may be worth simplifying this to all just be additive. I was thinking about it and this is just a temporary builder. To one of your earlier comments, I don't think we need to support removal of attributes and here it's hard to imagine the case for someone to do something like:

getTableBuilder(table)
  .setRequestAttribute("foo", "bar")
  .setRequestAttribute("abc", "def")
  .setRequestAttributes(Map.of("foo", "asdfa")) // actually forget those, i want to just use this
  .build()

I could see needing to support replacement/removal/etc if we allowed modifying the longer-lived Table objects. I know some builders out there have like a BuiltObject.toBuilder() method, in which case maybe it'd be good to support these things for evolving Tables over time. But our builders are pretty simple and one-off, so not necessary.

So trying to think about KISS here, and to me that means simply additive. I could even imagine changing the plural method to be named setAllRequestAttributes so its maybe a little clearer (along with javadoc) that we're just iterating and calling set on each. If anything, I could imagine a more useful bit of complexity to be supporting typed set methods, like setRequestAttribute(String, String) which does the byte[] conversion for you, along with other primitives. But we can skip that for now :)

One other note -- many people will not use request attributes. Let's default the requestAttributes map to null, and only instantiate a hashmap if one of these setters is called.

@bbeaudreault bbeaudreault Jul 20, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also, I think you need to update the ThriftConnection to override the new method.

@rmdmattingly rmdmattingly Jul 20, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍 good points, I'll update this morning


/**
* Create the {@link AsyncTable} instance.
*/
Expand Down
Loading