Skip to content

Commit 0170439

Browse files
shohamazonshohamazon
committed
Revert "fix(java): enforce immediate timeouts (valkey-io#5264)"
This reverts commit 512feef. Signed-off-by: Shoham Elias <shohame@amazon.com>
1 parent 404f4bd commit 0170439

File tree

14 files changed

+194
-813
lines changed

14 files changed

+194
-813
lines changed

‎java/client/src/main/java/glide/api/BaseClient.java‎

Lines changed: 20 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -500,9 +500,7 @@ protected static CommandManager buildCommandManager(ConnectionManager connection
500500
// We'll update this once the connection provides the native handle
501501
GlideCoreClient core =
502502
new GlideCoreClient(
503-
connectionManager.getNativeClientHandle(),
504-
connectionManager.getMaxInflightRequests(),
505-
connectionManager.getRequestTimeoutMs());
503+
connectionManager.getNativeClientHandle(), connectionManager.getMaxInflightRequests());
506504
// Register for PubSub push delivery
507505
try {
508506
GlideCoreClient.registerClient(connectionManager.getNativeClientHandle(), null);
@@ -2669,14 +2667,13 @@ public CompletableFuture<Map<GlideString, Double>> zpopmin(@NonNull GlideString
26692667
@Override
26702668
public CompletableFuture<Object[]> bzpopmin(@NonNull String[] keys, double timeout) {
26712669
String[] arguments = ArrayUtils.add(keys, Double.toString(timeout));
2672-
return commandManager.submitBlockingCommand(
2673-
BZPopMin, arguments, this::handleArrayOrNullResponse);
2670+
return commandManager.submitNewCommand(BZPopMin, arguments, this::handleArrayOrNullResponse);
26742671
}
26752672

26762673
@Override
26772674
public CompletableFuture<Object[]> bzpopmin(@NonNull GlideString[] keys, double timeout) {
26782675
GlideString[] arguments = ArrayUtils.add(keys, gs(Double.toString(timeout)));
2679-
return commandManager.submitBlockingCommand(
2676+
return commandManager.submitNewCommand(
26802677
BZPopMin, arguments, this::handleArrayOrNullResponseBinary);
26812678
}
26822679

@@ -2708,14 +2705,13 @@ public CompletableFuture<Map<GlideString, Double>> zpopmax(@NonNull GlideString
27082705
@Override
27092706
public CompletableFuture<Object[]> bzpopmax(@NonNull String[] keys, double timeout) {
27102707
String[] arguments = ArrayUtils.add(keys, Double.toString(timeout));
2711-
return commandManager.submitBlockingCommand(
2712-
BZPopMax, arguments, this::handleArrayOrNullResponse);
2708+
return commandManager.submitNewCommand(BZPopMax, arguments, this::handleArrayOrNullResponse);
27132709
}
27142710

27152711
@Override
27162712
public CompletableFuture<Object[]> bzpopmax(@NonNull GlideString[] keys, double timeout) {
27172713
GlideString[] arguments = ArrayUtils.add(keys, gs(Double.toString(timeout)));
2718-
return commandManager.submitBlockingCommand(
2714+
return commandManager.submitNewCommand(
27192715
BZPopMax, arguments, this::handleArrayOrNullResponseBinary);
27202716
}
27212717

@@ -3342,20 +3338,13 @@ public CompletableFuture<Map<GlideString, Map<GlideString, GlideString[][]>>> xr
33423338
public CompletableFuture<Map<String, Map<String, String[][]>>> xread(
33433339
@NonNull Map<String, String> keysAndIds, @NonNull StreamReadOptions options) {
33443340
String[] arguments = options.toArgs(keysAndIds);
3345-
if (options.isBlocking()) {
3346-
return commandManager.submitBlockingCommand(XRead, arguments, this::handleXReadResponse);
3347-
}
33483341
return commandManager.submitNewCommand(XRead, arguments, this::handleXReadResponse);
33493342
}
33503343

33513344
@Override
33523345
public CompletableFuture<Map<GlideString, Map<GlideString, GlideString[][]>>> xreadBinary(
33533346
@NonNull Map<GlideString, GlideString> keysAndIds, @NonNull StreamReadOptions options) {
33543347
GlideString[] arguments = options.toArgsBinary(keysAndIds);
3355-
if (options.isBlocking()) {
3356-
return commandManager.submitBlockingCommand(
3357-
XRead, arguments, this::handleXReadResponseBinary);
3358-
}
33593348
return commandManager.submitNewCommand(XRead, arguments, this::handleXReadResponseBinary);
33603349
}
33613350

@@ -3625,9 +3614,6 @@ public CompletableFuture<Map<String, Map<String, String[][]>>> xreadgroup(
36253614
@NonNull String consumer,
36263615
@NonNull StreamReadGroupOptions options) {
36273616
String[] arguments = options.toArgs(group, consumer, keysAndIds);
3628-
if (options.isBlocking()) {
3629-
return commandManager.submitBlockingCommand(XReadGroup, arguments, this::handleXReadResponse);
3630-
}
36313617
return commandManager.submitNewCommand(XReadGroup, arguments, this::handleXReadResponse);
36323618
}
36333619

@@ -3638,10 +3624,6 @@ public CompletableFuture<Map<GlideString, Map<GlideString, GlideString[][]>>> xr
36383624
@NonNull GlideString consumer,
36393625
@NonNull StreamReadGroupOptions options) {
36403626
GlideString[] arguments = options.toArgsBinary(group, consumer, keysAndIds);
3641-
if (options.isBlocking()) {
3642-
return commandManager.submitBlockingCommand(
3643-
XReadGroup, arguments, this::handleXReadResponseBinary);
3644-
}
36453627
return commandManager.submitNewCommand(XReadGroup, arguments, this::handleXReadResponseBinary);
36463628
}
36473629

@@ -4124,14 +4106,14 @@ public CompletableFuture<Long> linsert(
41244106
@Override
41254107
public CompletableFuture<String[]> blpop(@NonNull String[] keys, double timeout) {
41264108
String[] arguments = ArrayUtils.add(keys, Double.toString(timeout));
4127-
return commandManager.submitBlockingCommand(
4109+
return commandManager.submitNewCommand(
41284110
BLPop, arguments, response -> castArray(handleArrayOrNullResponse(response), String.class));
41294111
}
41304112

41314113
@Override
41324114
public CompletableFuture<GlideString[]> blpop(@NonNull GlideString[] keys, double timeout) {
41334115
GlideString[] arguments = ArrayUtils.add(keys, gs(Double.toString(timeout)));
4134-
return commandManager.submitBlockingCommand(
4116+
return commandManager.submitNewCommand(
41354117
BLPop,
41364118
arguments,
41374119
response -> castArray(handleArrayOrNullResponseBinary(response), GlideString.class));
@@ -4140,14 +4122,14 @@ public CompletableFuture<GlideString[]> blpop(@NonNull GlideString[] keys, doubl
41404122
@Override
41414123
public CompletableFuture<String[]> brpop(@NonNull String[] keys, double timeout) {
41424124
String[] arguments = ArrayUtils.add(keys, Double.toString(timeout));
4143-
return commandManager.submitBlockingCommand(
4125+
return commandManager.submitNewCommand(
41444126
BRPop, arguments, response -> castArray(handleArrayOrNullResponse(response), String.class));
41454127
}
41464128

41474129
@Override
41484130
public CompletableFuture<GlideString[]> brpop(@NonNull GlideString[] keys, double timeout) {
41494131
GlideString[] arguments = ArrayUtils.add(keys, gs(Double.toString(timeout)));
4150-
return commandManager.submitBlockingCommand(
4132+
return commandManager.submitNewCommand(
41514133
BRPop,
41524134
arguments,
41534135
response -> castArray(handleArrayOrNullResponseBinary(response), GlideString.class));
@@ -4306,7 +4288,7 @@ public CompletableFuture<Map<String, Object>> bzmpop(
43064288
new String[] {Double.toString(timeout), Integer.toString(keys.length)},
43074289
keys,
43084290
new String[] {modifier.toString()});
4309-
return commandManager.submitBlockingCommand(
4291+
return commandManager.submitNewCommand(
43104292
BZMPop,
43114293
arguments,
43124294
response -> convertKeyValueArrayToMap(handleArrayOrNullResponse(response), Double.class));
@@ -4320,7 +4302,7 @@ public CompletableFuture<Map<GlideString, Object>> bzmpop(
43204302
new GlideString[] {gs(Double.toString(timeout)), gs(Integer.toString(keys.length))},
43214303
keys,
43224304
new GlideString[] {gs(modifier.toString())});
4323-
return commandManager.submitBlockingCommand(
4305+
return commandManager.submitNewCommand(
43244306
BZMPop,
43254307
arguments,
43264308
response ->
@@ -4336,7 +4318,7 @@ public CompletableFuture<Map<String, Object>> bzmpop(
43364318
new String[] {Double.toString(timeout), Integer.toString(keys.length)},
43374319
keys,
43384320
new String[] {modifier.toString(), COUNT_VALKEY_API, Long.toString(count)});
4339-
return commandManager.submitBlockingCommand(
4321+
return commandManager.submitNewCommand(
43404322
BZMPop,
43414323
arguments,
43424324
response -> convertKeyValueArrayToMap(handleArrayOrNullResponse(response), Double.class));
@@ -4352,7 +4334,7 @@ public CompletableFuture<Map<GlideString, Object>> bzmpop(
43524334
new GlideString[] {
43534335
gs(modifier.toString()), gs(COUNT_VALKEY_API), gs(Long.toString(count))
43544336
});
4355-
return commandManager.submitBlockingCommand(
4337+
return commandManager.submitNewCommand(
43564338
BZMPop,
43574339
arguments,
43584340
response ->
@@ -4750,7 +4732,7 @@ public CompletableFuture<Map<String, String[]>> blmpop(
47504732
new String[] {Double.toString(timeout), Long.toString(keys.length)},
47514733
keys,
47524734
new String[] {direction.toString(), COUNT_FOR_LIST_VALKEY_API, Long.toString(count)});
4753-
return commandManager.submitBlockingCommand(
4735+
return commandManager.submitNewCommand(
47544736
BLMPop,
47554737
arguments,
47564738
response -> castMapOfArrays(handleMapOrNullResponse(response), String.class));
@@ -4766,7 +4748,7 @@ public CompletableFuture<Map<GlideString, GlideString[]>> blmpop(
47664748
new GlideString[] {
47674749
gs(direction.toString()), gs(COUNT_FOR_LIST_VALKEY_API), gs(Long.toString(count))
47684750
});
4769-
return commandManager.submitBlockingCommand(
4751+
return commandManager.submitNewCommand(
47704752
BLMPop,
47714753
arguments,
47724754
response ->
@@ -4782,7 +4764,7 @@ public CompletableFuture<Map<String, String[]>> blmpop(
47824764
new String[] {Double.toString(timeout), Long.toString(keys.length)},
47834765
keys,
47844766
new String[] {direction.toString()});
4785-
return commandManager.submitBlockingCommand(
4767+
return commandManager.submitNewCommand(
47864768
BLMPop,
47874769
arguments,
47884770
response -> castMapOfArrays(handleMapOrNullResponse(response), String.class));
@@ -4796,7 +4778,7 @@ public CompletableFuture<Map<GlideString, GlideString[]>> blmpop(
47964778
new GlideString[] {gs(Double.toString(timeout)), gs(Long.toString(keys.length))},
47974779
keys,
47984780
new GlideString[] {gs(direction.toString())});
4799-
return commandManager.submitBlockingCommand(
4781+
return commandManager.submitNewCommand(
48004782
BLMPop,
48014783
arguments,
48024784
response ->
@@ -4850,8 +4832,7 @@ public CompletableFuture<String> blmove(
48504832
new String[] {
48514833
source, destination, wherefrom.toString(), whereto.toString(), Double.toString(timeout)
48524834
};
4853-
return commandManager.submitBlockingCommand(
4854-
BLMove, arguments, this::handleStringOrNullResponse);
4835+
return commandManager.submitNewCommand(BLMove, arguments, this::handleStringOrNullResponse);
48554836
}
48564837

48574838
@Override
@@ -4869,7 +4850,7 @@ public CompletableFuture<GlideString> blmove(
48694850
gs(whereto.toString()),
48704851
gs(Double.toString(timeout))
48714852
};
4872-
return commandManager.submitBlockingCommand(
4853+
return commandManager.submitNewCommand(
48734854
BLMove, arguments, this::handleGlideStringOrNullResponse);
48744855
}
48754856

@@ -5976,7 +5957,7 @@ public CompletableFuture<Object[]> hscan(
59765957

59775958
@Override
59785959
public CompletableFuture<Long> wait(long numreplicas, long timeout) {
5979-
return commandManager.submitBlockingCommand(
5960+
return commandManager.submitNewCommand(
59805961
Wait,
59815962
new String[] {Long.toString(numreplicas), Long.toString(timeout)},
59825963
this::handleLongResponse);

‎java/client/src/main/java/glide/api/GlideClient.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import static command_request.CommandRequestOuterClass.RequestType.ConfigResetStat;
88
import static command_request.CommandRequestOuterClass.RequestType.ConfigRewrite;
99
import static command_request.CommandRequestOuterClass.RequestType.ConfigSet;
10+
import static command_request.CommandRequestOuterClass.RequestType.CustomCommand;
1011
import static command_request.CommandRequestOuterClass.RequestType.DBSize;
1112
import static command_request.CommandRequestOuterClass.RequestType.Echo;
1213
import static command_request.CommandRequestOuterClass.RequestType.FlushAll;
@@ -134,12 +135,13 @@ public static CompletableFuture<GlideClient> createClient(
134135

135136
@Override
136137
public CompletableFuture<Object> customCommand(@NonNull String[] args) {
137-
return commandManager.submitCustomCommand(args, this::handleObjectOrNullResponse);
138+
return commandManager.submitNewCommand(CustomCommand, args, this::handleObjectOrNullResponse);
138139
}
139140

140141
@Override
141142
public CompletableFuture<Object> customCommand(@NonNull GlideString[] args) {
142-
return commandManager.submitCustomCommand(args, this::handleBinaryObjectOrNullResponse);
143+
return commandManager.submitNewCommand(
144+
CustomCommand, args, this::handleBinaryObjectOrNullResponse);
143145
}
144146

145147
@Deprecated

‎java/client/src/main/java/glide/api/GlideClusterClient.java‎

Lines changed: 12 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import static command_request.CommandRequestOuterClass.RequestType.ConfigResetStat;
88
import static command_request.CommandRequestOuterClass.RequestType.ConfigRewrite;
99
import static command_request.CommandRequestOuterClass.RequestType.ConfigSet;
10+
import static command_request.CommandRequestOuterClass.RequestType.CustomCommand;
1011
import static command_request.CommandRequestOuterClass.RequestType.DBSize;
1112
import static command_request.CommandRequestOuterClass.RequestType.Echo;
1213
import static command_request.CommandRequestOuterClass.RequestType.FCall;
@@ -164,29 +165,31 @@ public static CompletableFuture<GlideClusterClient> createClient(
164165
@Override
165166
public CompletableFuture<ClusterValue<Object>> customCommand(@NonNull String[] args) {
166167
// TODO if a command returns a map as a single value, ClusterValue misleads user
167-
return commandManager.submitCustomCommand(
168-
args, response -> ClusterValue.of(handleObjectOrNullResponse(response)));
168+
return commandManager.submitNewCommand(
169+
CustomCommand, args, response -> ClusterValue.of(handleObjectOrNullResponse(response)));
169170
}
170171

171172
@Override
172173
public CompletableFuture<ClusterValue<Object>> customCommand(@NonNull GlideString[] args) {
173174
// TODO if a command returns a map as a single value, ClusterValue misleads user
174-
return commandManager.submitCustomCommand(
175-
args, response -> ClusterValue.of(handleBinaryObjectOrNullResponse(response)));
175+
return commandManager.submitNewCommand(
176+
CustomCommand,
177+
args,
178+
response -> ClusterValue.of(handleBinaryObjectOrNullResponse(response)));
176179
}
177180

178181
@Override
179182
public CompletableFuture<ClusterValue<Object>> customCommand(
180183
@NonNull String[] args, @NonNull Route route) {
181-
return commandManager.submitCustomCommand(
182-
args, route, response -> handleCustomCommandResponse(route, response));
184+
return commandManager.submitNewCommand(
185+
CustomCommand, args, route, response -> handleCustomCommandResponse(route, response));
183186
}
184187

185188
@Override
186189
public CompletableFuture<ClusterValue<Object>> customCommand(
187190
@NonNull GlideString[] args, @NonNull Route route) {
188-
return commandManager.submitCustomCommand(
189-
args, route, response -> handleCustomCommandBinaryResponse(route, response));
191+
return commandManager.submitNewCommand(
192+
CustomCommand, args, route, response -> handleCustomCommandBinaryResponse(route, response));
190193
}
191194

192195
@SuppressWarnings("unchecked")
@@ -1304,7 +1307,7 @@ public CompletableFuture<GlideString> randomKeyBinary() {
13041307
@Override
13051308
public CompletableFuture<Long> wait(long numreplicas, long timeout) {
13061309
String[] arguments = new String[] {Long.toString(numreplicas), Long.toString(timeout)};
1307-
return commandManager.submitBlockingCommand(
1310+
return commandManager.submitNewCommand(
13081311
Wait, arguments, SimpleSingleNodeRoute.RANDOM, this::handleLongResponse);
13091312
}
13101313

‎java/client/src/main/java/glide/api/models/commands/stream/StreamReadOptions.java‎

Lines changed: 0 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -36,20 +36,6 @@ public class StreamReadOptions {
3636
*/
3737
protected Long count;
3838

39-
/**
40-
* Returns true if this options object specifies a BLOCK timeout, making the command a blocking
41-
* command.
42-
*
43-
* <p>Note: We check {@code block != null} rather than {@code block != null && block != 0} because
44-
* {@code BLOCK 0} means "block indefinitely" in Valkey/Redis, which is still a blocking command
45-
* that should skip Java-side timeout enforcement.
46-
*
47-
* @return true if BLOCK option is set (including BLOCK 0 for indefinite blocking)
48-
*/
49-
public boolean isBlocking() {
50-
return this.block != null;
51-
}
52-
5339
/**
5440
* Converts options and the key-to-id input for {@link StreamBaseCommands#xread(Map,
5541
* StreamReadOptions)} into a String[].

‎java/client/src/main/java/glide/ffi/resolvers/NativeUtils.java‎

Lines changed: 5 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ public class NativeUtils {
2929
public static final String NATIVE_FOLDER_PATH_PREFIX = "nativeutils";
3030

3131
/** Temporary directory which will contain the dynamic library files. */
32-
private static volatile File temporaryDir;
32+
private static File temporaryDir;
3333

3434
/** Track if the Glide library has already been loaded */
3535
private static volatile boolean glideLibLoaded = false;
@@ -95,17 +95,12 @@ public static void loadLibraryFromJar(String path) throws IOException {
9595
}
9696

9797
// Prepare temporary file
98-
File localTempDir;
99-
synchronized (NativeUtils.class) {
100-
if (temporaryDir == null) {
101-
File createdDir = createTempDirectory(NATIVE_FOLDER_PATH_PREFIX);
102-
createdDir.deleteOnExit();
103-
temporaryDir = createdDir;
104-
}
105-
localTempDir = temporaryDir;
98+
if (temporaryDir == null) {
99+
temporaryDir = createTempDirectory(NATIVE_FOLDER_PATH_PREFIX);
100+
temporaryDir.deleteOnExit();
106101
}
107102

108-
File temp = new File(localTempDir, filename);
103+
File temp = new File(temporaryDir, filename);
109104

110105
try (InputStream is = NativeUtils.class.getResourceAsStream(path)) {
111106
if (is == null) {

0 commit comments

Comments
 (0)