scottf

import io.nats.client.*;
import io.nats.client.api.*;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;

scottf

internal static class KvSubjectTransform
{
    static string kvName1 = "stest";
    static string kvName2 = "stest-invoices";
    
    static void Main(string[] args)
    {
        try
        {
            Options opts = ConnectionFactory.GetDefaultOptions();

scottf

import io.nats.client.*;
import io.nats.client.api.*;
import io.nats.client.impl.NatsJetStreamMetaData;
import io.nats.client.support.Status;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class RawFetchNoWaitPlusExpires {

scottf

class Base
{
    public virtual void Foo()
    {
        Console.WriteLine("Base.Foo from " + GetType());
    }
}

class Newit : Base
{

scottf

package io.nats.examples.jetstream;

import io.nats.client.*;
import io.nats.client.api.*;

import java.io.IOException;

public class PullInactiveChecks {

    static String STREAM = "InactiveStream";

scottf

I also built a tuner program and have these numbers. I seeded the stream with 100,000 x 50 or x 8000 byte messages. I ran 10 rounds of every test type (randomizing the order of types for each round) so each type ended up reading 1,000,000 messages. This is against a cluster and the stream is R3

My takeaway is that bigger batches do better on these long continuous pulls. I didn't include batches of 1 or 10 because they were so much slower. For the reader, I realized that for smaller batches I need to repull sooner, but bigger batches 80% seems good enough.

50 byte message size and 8000 byte message size

Type	Batch	50 bytes	Elapsed ms	msg/ms	ms/msg	8000 bytes	Elapsed ms	msg/ms	ms/msg
Fetch	50		9599 ms	104.18	0.0096		16417 ms	60.91	0.0164
Fetch	100		8322 ms	120.16	0.0083		15656 ms	63.87	0.0157
Fetch	250

scottf

import io.nats.client.Connection;
import io.nats.client.JetStreamApiException;
import io.nats.client.JetStreamManagement;
import io.nats.client.Nats;
import io.nats.client.api.DiscardPolicy;
import io.nats.client.api.StreamConfiguration;
import io.nats.client.api.StreamInfo;
import io.nats.client.support.NatsKeyValueUtil;

import java.util.List;

scottf

@Test
public void testOneShot() throws Exception {

    runInJsServer(connection -> {

        JetStream js = connection.jetStream();
        JetStreamManagement jsm = connection.jetStreamManagement();

        // Create the stream.
        StreamConfiguration sc = StreamConfiguration.builder()

scottf

import io.nats.client.*;
import io.nats.client.api.ConsumerConfiguration;
import io.nats.client.api.StorageType;
import io.nats.client.api.StreamConfiguration;

import java.io.IOException;
import java.time.Duration;
import java.util.List;

class FetchTests {

scottf

# Copyright 2022 The NATS Authors
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.