Fix append failures caused by invalid latency metrics

metadata/src/main/java/io/streamstack/metadata/raft/MetadataClient.java

@@ -7,6 +7,7 @@
 import com.alipay.sofa.jraft.closure.ReadIndexClosure;
 import com.alipay.sofa.jraft.conf.Configuration;
 import com.alipay.sofa.jraft.entity.PeerId;
+import com.alipay.sofa.jraft.error.RaftError;
 import com.alipay.sofa.jraft.option.CliOptions;
 import com.alipay.sofa.jraft.rpc.impl.cli.CliClientServiceImpl;
 
@@ -181,37 +182,62 @@ public CompletableFuture<Void> readIndex(Runnable read) {
 
  public <T> CompletableFuture<T> readIndex(Supplier<T> read) {
  CompletableFuture<T> future = new CompletableFuture<>();
+ attemptReadIndex(read, 0, future);
+ return future;
+ }
 
- node.raftNode().readIndex(null, new ReadIndexClosure() {
- @Override
- public void run(Status status, long index, byte[] reqCtx) {
- if (!status.isOk()) {
- future.completeExceptionally(new IllegalStateException(status.getErrorMsg()));
- return;
- }
+ private <T> void attemptReadIndex(Supplier<T> read, int attempt, CompletableFuture<T> future) {
+ if (future.isDone()) {
+ return;
+ }
 
- node.stateMachine().awaitApplied(index).whenCompleteAsync((ignored, error) -> {
- if (Objects.nonNull(error)) {
- future.completeExceptionally(unwrap(error));
+ try {
+ node.raftNode().readIndex(null, new ReadIndexClosure() {
+ @Override
+ public void run(Status status, long index, byte[] reqCtx) {
+ if (!status.isOk()) {
+ if (status.getRaftError() == RaftError.EAGAIN && attempt < maxRetries) {
+ scheduleReadIndexRetry(read, attempt, future);
+ } else {
+ future.completeExceptionally(new IllegalStateException(status.getErrorMsg()));
+ }
  return;
  }
 
- try {
- future.complete(node.stateMachine().read(read));
- } catch (Throwable t) {
- Throwable cause = unwrap(t);
+ node.stateMachine().awaitApplied(index).whenCompleteAsync((ignored, error) -> {
+ if (Objects.nonNull(error)) {
+ future.completeExceptionally(unwrap(error));
+ return;
+ }
 
- if (cause instanceof MetadataException metadataException) {
- future.completeExceptionally(MetadataException.toStreamClientException(metadataException));
- } else {
- future.completeExceptionally(cause);
+ try {
+ future.complete(node.stateMachine().read(read));
+ } catch (Throwable t) {
+ Throwable cause = unwrap(t);
+
+ if (cause instanceof MetadataException metadataException) {
+ future.completeExceptionally(MetadataException.toStreamClientException(metadataException));
+ } else {
+ future.completeExceptionally(cause);
+ }
  }
- }
- }, scheduler);
- }
- });
+ }, scheduler);
+ }
+ });
+ } catch (Throwable t) {
+ future.completeExceptionally(unwrap(t));
+ }
+ }
 
- return future;
+ private <T> void scheduleReadIndexRetry(Supplier<T> read, int attempt, CompletableFuture<T> future) {
+ try {
+ scheduler.schedule(
+ () -> attemptReadIndex(read, attempt + 1, future),
+ retrySleepMs,
+ TimeUnit.MILLISECONDS);
+ } catch (RuntimeException e) {
+ future.completeExceptionally(e);
+ }
  }
 
  @Override

stream/src/main/java/io/streamstack/s3/S3Storage.java

@@ -334,7 +334,11 @@ public CompletableFuture<Void> append(AppendContext context, StreamRecordBatch s
  append0(context, writeRequest, false);
  return cf.whenComplete((nil, ex) -> {
  streamRecord.release();
- APPEND_STORAGE_LATENCY.record(TimerUtil.timeElapsedSince(startTime, TimeUnit.NANOSECONDS));
+ try {
+ APPEND_STORAGE_LATENCY.record(TimerUtil.timeElapsedSince(startTime, TimeUnit.NANOSECONDS));
+ } catch (RuntimeException metricsError) {
+ LOGGER.warn("Failed to record append storage latency", metricsError);
+ }
  });
  }
 

stream/src/main/java/io/streamstack/s3/S3Stream.java

@@ -243,7 +243,11 @@ public CompletableFuture<AppendResult> append(AppendContext context, RecordBatch
  pendingAppends.add(cf);
  PendingRequestTracker.Handle pendingAppend = PENDING_APPEND_TRACKER.begin();
  return cf.whenComplete((nil, ex) -> {
- APPEND_STREAM_LATENCY.record(TimerUtil.timeElapsedSince(startTimeNanos, TimeUnit.NANOSECONDS));
+ try {
+ APPEND_STREAM_LATENCY.record(TimerUtil.timeElapsedSince(startTimeNanos, TimeUnit.NANOSECONDS));
+ } catch (RuntimeException metricsError) {
+ logger.warn("Failed to record append stream latency", metricsError);
+ }
  pendingAppends.remove(cf);
  pendingAppend.close();
  });

stream/src/main/java/io/streamstack/s3/metrics/wrapper/DeltaHistogram.java

@@ -103,6 +103,10 @@ private void update(long candidate, AtomicLong target, BiPredicate<Long, Long> p
  }
 
  public void record(long value) {
+ // Silently drop invalid samples because HdrHistogram only supports non-negative values.
+ if (value < 0) {
+ return;
+ }
  cumulativeCount.increment();
  cumulativeSum.add(value);
  this.recorder.recordValue(value);

stream/src/test/java/io/streamstack/s3/metrics/wrapper/MetricsWrapperTest.java

@@ -112,6 +112,19 @@ public void testDeltaHistogram() throws InterruptedException {
 // Assertions.assertEquals(15000, p50, 1000);
  }
 
+ @Test
+ public void testDeltaHistogramIgnoresNegativeValues() {
+ DeltaHistogram histogram = new DeltaHistogram();
+
+ Assertions.assertDoesNotThrow(() -> histogram.record(-1));
+ Assertions.assertEquals(0, histogram.cumulativeCount());
+ Assertions.assertEquals(0, histogram.cumulativeSum());
+
+ histogram.record(1);
+ Assertions.assertEquals(1, histogram.cumulativeCount());
+ Assertions.assertEquals(1, histogram.cumulativeSum());
+ }
+
  private void mockLinearDataDist(DeltaHistogram histogram, int init, int steps) {
  for (int i = init; i < init + steps; i++) {
  histogram.record(i);