HDDS-14036. StreamRead: Make preReadSize, responseDataSize and read timeout configurable#9461
HDDS-14036. StreamRead: Make preReadSize, responseDataSize and read timeout configurable#9461szetszwo merged 11 commits intoapache:masterfrom
Conversation
|
|
||
| ByteBuffer readFromQueue() throws IOException { | ||
| final ReadBlockResponseProto readBlock = poll(10, TimeUnit.SECONDS); | ||
| final ReadBlockResponseProto readBlock = poll(readTimeoutMs, TimeUnit.SECONDS); |
There was a problem hiding this comment.
Thanks @Russole for working on this, left few comments
Because readTimeoutMs is in milliseconds, you must pass TimeUnit.MILLISECONDS so the timeout isn’t misinterpreted as seconds
| final ReadBlockResponseProto readBlock = poll(readTimeoutMs, TimeUnit.SECONDS); | |
| final ReadBlockResponseProto readBlock = poll(readTimeoutMs, TimeUnit.MILLISECONDS); |
| private int streamReadResponseDataSize = 1 << 20; | ||
|
|
||
| @Config(key = "ozone.client.stream.read.timeout", | ||
| defaultValue = "1000", |
There was a problem hiding this comment.
The default value should be 10000 in order to match with the ozone-default.xml
adoroszlai
left a comment
There was a problem hiding this comment.
Thanks @Russole for working on this.
Please don't also add the config to OzoneConfigKeys and ozone-default.xml.
Please add test cases to verify custom config is applied.
| defaultValue = "1000", | ||
| type = ConfigType.INT, |
There was a problem hiding this comment.
Please use ConfigType.TIME and "10s".
| type = ConfigType.INT, | ||
| tags = {ConfigTag.CLIENT}, | ||
| description = "Timeout in ms for receiving streaming read responses.") | ||
| private int streamReadTimeoutMs = 10_000; |
There was a problem hiding this comment.
Plese use Duration instead of int and drop Ms from name.
| public int getStreamReadTimeoutMs() { | ||
| return streamReadTimeoutMs; |
There was a problem hiding this comment.
Return Duration and drop Ms from name.
|
Many thanks to @adoroszlai, @sreejasahithi, and @jojochuang for the reviews. I've addressed the comments — please let me know if any further changes are needed. |
adoroszlai
left a comment
There was a problem hiding this comment.
Thanks @Russole for updating the patch.
| public static final String | ||
| OZONE_CLIENT_STREAM_READ_PRE_READ_SIZE = | ||
| "ozone.client.stream.read.pre-read-size"; | ||
| public static final long | ||
| OZONE_CLIENT_STREAM_READ_PRE_READ_SIZE_DEFAULT = | ||
| 32L << 20; | ||
|
|
||
| public static final String | ||
| OZONE_CLIENT_STREAM_READ_RESPONSE_DATA_SIZE = | ||
| "ozone.client.stream.read.response-data-size"; | ||
| public static final int | ||
| OZONE_CLIENT_STREAM_READ_RESPONSE_DATA_SIZE_DEFAULT = | ||
| 1 << 20; | ||
|
|
||
| public static final String | ||
| OZONE_CLIENT_STREAM_READ_TIMEOUT = | ||
| "ozone.client.stream.read.timeout"; | ||
| public static final int | ||
| OZONE_CLIENT_STREAM_READ_TIMEOUT_DEFAULT = | ||
| 10_000; | ||
|
|
There was a problem hiding this comment.
Sorry, if my original comment was confusing. Please remove these, too.
| public static final String | |
| OZONE_CLIENT_STREAM_READ_PRE_READ_SIZE = | |
| "ozone.client.stream.read.pre-read-size"; | |
| public static final long | |
| OZONE_CLIENT_STREAM_READ_PRE_READ_SIZE_DEFAULT = | |
| 32L << 20; | |
| public static final String | |
| OZONE_CLIENT_STREAM_READ_RESPONSE_DATA_SIZE = | |
| "ozone.client.stream.read.response-data-size"; | |
| public static final int | |
| OZONE_CLIENT_STREAM_READ_RESPONSE_DATA_SIZE_DEFAULT = | |
| 1 << 20; | |
| public static final String | |
| OZONE_CLIENT_STREAM_READ_TIMEOUT = | |
| "ozone.client.stream.read.timeout"; | |
| public static final int | |
| OZONE_CLIENT_STREAM_READ_TIMEOUT_DEFAULT = | |
| 10_000; |
|
|
||
| Token<OzoneBlockTokenIdentifier> token = null; | ||
| // Mock XceiverClientFactory since StreamBlockInputStream requires it in the constructor | ||
| XceiverClientFactory xceiverClientFactory = Mockito.mock(XceiverClientFactory.class); |
There was a problem hiding this comment.
nit: please add import static org.mockito.Mockito.mock;
| conf.set("ozone.client.stream.read.pre-read-size", "67108864"); | ||
| conf.set("ozone.client.stream.read.response-data-size", "2097152"); | ||
| conf.set("ozone.client.stream.read.timeout", "5s"); | ||
|
|
||
| OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class); |
There was a problem hiding this comment.
OzoneConfiguration is not required, please create OzoneClientConfig directly and set values using its methods.
OzoneClientConfig clientConfig = new OzoneClientConfig();
clientConfig.set...(...);| } | ||
|
|
||
| @Test | ||
| public void testCustomStreamReadConfigIsApplied() throws Exception { |
There was a problem hiding this comment.
This should be a unit test, since it does not require a working cluster. Please move to new test class hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java.
| OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class); | ||
|
|
||
| // Sanity check | ||
| assertEquals(Duration.ofSeconds(5), clientConfig.getStreamReadTimeout()); |
There was a problem hiding this comment.
To test whether OzoneClientConfig applies the new config keys correctly, add test case in existing TestOzoneClientConfig.
| // Convert Duration -> int seconds for poll(...) | ||
| final int timeoutSeconds; | ||
| if (readTimeout == null || readTimeout.isZero() || readTimeout.isNegative()) { | ||
| timeoutSeconds = 0; | ||
| } else { | ||
| long sec = readTimeout.getSeconds(); | ||
| // Prevent overflow if client config is extremely large | ||
| timeoutSeconds = sec > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) sec; | ||
| } | ||
| final ReadBlockResponseProto readBlock = poll(timeoutSeconds, TimeUnit.SECONDS); |
There was a problem hiding this comment.
- Conversion to seconds is unnecessary, since
pollthen converts to nanoseconds. - Validation should be done in
OzoneClientConfig#validate() - Calculate
timeoutNanosinStreamBlockInputStreamconstructor asconfig.getStreamReadTimeout().toNanos().pollparameters can be removed.
| private final int responseDataSize; // Default size is 1 MB | ||
| private final long preReadSize; // Default size is 32 MB | ||
| private final Duration readTimeout; // // Default timeout is 10 second |
There was a problem hiding this comment.
Please remove the comments, they easily get outdated. Default values can be checked in the config.
adoroszlai
left a comment
There was a problem hiding this comment.
Thanks @Russole for updating the patch. LGTM, except two minor items.
| ReadBlockResponseProto poll(int timeout, TimeUnit timeoutUnit) throws IOException { | ||
| final long timeoutNanos = timeoutUnit.toNanos(timeout); | ||
| ReadBlockResponseProto poll() throws IOException { | ||
| final long timeoutNanos = readTimeoutNanos; |
There was a problem hiding this comment.
nit: remove timeoutNanos and use readTimeoutNanos directly.
| "ozone.client.elastic.byte.buffer.pool.max.size"; | ||
| public static final String OZONE_CLIENT_ELASTIC_BYTE_BUFFER_POOL_MAX_SIZE_DEFAULT = "16GB"; | ||
|
|
||
There was a problem hiding this comment.
nit: please avoid whiitespace-only change
sreejasahithi
left a comment
There was a problem hiding this comment.
Thanks @Russole for working on this.
LGTM
|
Thanks @Russole for updating the patch. Please note that the whitespace change was made in the wrong line, so now we have 2 changed lines instead of 0. There should be no changes in |
|
Thank you @adoroszlai for the reminder. I have removed all whitespace-only changes from hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java. |
|
CI failures look flaky and unrelated to this change. |
szetszwo
left a comment
There was a problem hiding this comment.
+1 the change looks good.
|
@Russole , thanks for working on this! @adoroszlai , thanks for reviewing this! |
What changes were proposed in this pull request?
This change makes the fixed values used by StreamBlockInputStream for
preReadSize, responseDataSize, and the streaming read timeout configurable
via OzoneClientConfig and ozone-default.xml.
This allows users and operators to tune stream-read performance based on workload
characteristics and cluster latency.
Currently, StreamBlockInputStream uses the following fixed values:
These defaults work for general cases but may not be optimal for all environments, particularly under higher latency conditions or workloads that benefit from larger streaming buffers or more fine-grained timeout control.
This PR introduces three new client-side configuration keys to make these parameters fully configurable:
Changes include:
What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/HDDS-14036
How was this patch tested?
GitHub Actions CI for my fork ran successfully with all checks passing.