Through FluxBufferTimeout through BufferTimeoutWithBackpressureSubscriber:
|
this.prefetch = batchSize << 2; |
|
this.replenishMark = batchSize << 1; |
If batchSize (maxSize in bufferTimeout) is very big, prefetch and replenishMark can become negative. In my experience, this can cause a thread to silently hang. I would expect a large input to gracefully saturate these internal variables or at least signal an error.
Integer.MAX_VALUE would work well for ‘unbounded’ buffering if not for this; the brittle-looking Integer.MAX_VALUE >> 2 will do for now.
Versions affected: commits since PR #3332 as far as I know. Link above represents main as of time of writing.
Through
FluxBufferTimeoutthroughBufferTimeoutWithBackpressureSubscriber:reactor-core/reactor-core/src/main/java/reactor/core/publisher/FluxBufferTimeout.java
Lines 215 to 216 in c431f54
If
batchSize(maxSizeinbufferTimeout) is very big,prefetchandreplenishMarkcan become negative. In my experience, this can cause a thread to silently hang. I would expect a large input to gracefully saturate these internal variables or at least signal an error.Integer.MAX_VALUEwould work well for ‘unbounded’ buffering if not for this; the brittle-lookingInteger.MAX_VALUE >> 2will do for now.Versions affected: commits since PR #3332 as far as I know. Link above represents
mainas of time of writing.