Best practices for Inter Step Buffer (Nats JetStream)?

[qianbeibuzui]

My Inter Step Buffer (Nats JetStream) and vertices settings are shown in the figure:

I have three doubts here:
What is the relationship between JetStream.maxMss: 50000 and verticals.limits.bufferMaxLength: 30000? If I set two vertices in the Pipeline with a buffer MaxLength of 30000, will there be a bug? What is the best practice plan?

2. JetStream.max_cemory_store: 1073741824, I set 1Gi of memory here, but the actual pod is occupying 4GB of memory. What is the reason for this? And when preparing to break through 4G memory, it seems to automatically release memory? As shown in the picture:

3.What is the best practice for memory and maxDLLs for Inter Step Buffer (Nats JetStream)?

[kohlisid]

Hey @qianbeibuzui
I will try to cover your queries here!

1. JetStream.maxMss: 50000 and verticals.limits.bufferMaxLength: 30000? If I set two vertices in the Pipeline with a buffer MaxLength of 30000

Jetstream maxMsgs are the limit on the number of messages that can be stored in a stream, this will apply as an upper bound. Beyond these limits the discard policy kicks in and the messages are deleted/discarded accordingly.

bufferMaxLength is a soft limit getting enforced at the Numaflow level, where as a platform we try to not write more than this number to the buffer. And if the availability crosses the threshold, the user can specifify what action they want to take. Default is Numaflow retrying until success.

So there is no bug, Numaflow will try to honor the bufferMaxLength first. JetStream.maxMss is honored at a stream level by Jetstream.

2. max_memory_store
   max_memory_store | Maximum size of the 'memory' storage | Default is 75% of available memory

This is the Jetstream server resource limit that is used, if you are using memory mode they streams will be stored in memory.

Also for your automatically release memory
This could partly be a behavior from the golang GC. Could you let us know whats the GOMEMLIMIT getting used in your server.

3. What is the best practice for memory and maxDLLs for Inter Step Buffer (Nats JetStream)?
   To answer this I would need to understand a bit more of your use case, TPS and you data/msg size

I see that you are using stream replication as 1 with in memory mode. This makes it highly temporal
Any reason for this specifically?

For resiliency we should be using a File mode, along with a higher replication factor.
Typical usage to operate in typical outages and balance performance would be a file-based stream with R=3.

[qianbeibuzui]

1.JetStream.maxMss: 50000 and verticals.limits.bufferMaxLength: 30000 * 2. 60000>50000.If Numaflow's cache reaches 30000 * 2, will messages be lost due to JetStream. maxMax: 50000?

2.you are using stream replication as 1 with in memory mode. This makes it highly temporal
Any reason for this specifically?
It's because I want to pursue higher TPS.

[vigith]

It's because I want to pursue higher TPS.

What is the TPS you are trying to hit? Also, are you doing reduce operations also in the pipeline?

[qianbeibuzui]

因为我想追求更高的TPS。

您要达到的 TPS 是多少？此外，您是否也在管道中执行减​​少操作？

1W qps。But I have noticed that the official update for MonoVertex has been released, but my attempt to use MonoVertex.source.kafka has failed. It seems that MonoVertex does not have a source. kafka. This requires either official implementation or my own implementation of udsource.kafka.
Also, can you answer the first question I described?

Also, can you answer the first question I described?

JetStream.maxMss: 50000 且verticals.limits.bufferMaxLength: 30000 * 2. 60000>50000.如果Numaflow的缓存达到了30000 * 2，会不会因为JetStream.maxMax: 50000而丢失消息？

[qianbeibuzui]

It's because I want to pursue higher TPS.

What is the TPS you are trying to hit? Also, are you doing reduce operations also in the pipeline?

Also, are you doing reduce operations also in the pipeline?

It's not working yet. But I have a premonition: it is highly likely to be used in the future.

[vigith]

But I have noticed that the official update for MonoVertex has been released, but my attempt to use MonoVertex.source.kafka has failed. It seems that MonoVertex does not have a source. kafka. This requires either official implementation or my own implementation of udsource.kafka.

That is right. MonoVertex does not support builtin Kafka yet due to some breaking changes. We will support it in 1.5 release (approx ETA Feb/01/2025).

[vigith]

What is the relationship between JetStream.maxMss: 50000 and verticals.limits.bufferMaxLength: 30000? If I set two vertices in the Pipeline with a buffer MaxLength of 30000, will there be a bug? What is the best practice plan?

Each vertex has its own buffer, so each vertex will get independent buffers with independent limits. You can have 2 vertices without any problems.

[qianbeibuzui]

I'm very sorry to bother you again. I seem to have encountered a memory leak situation.
I continue to conduct stress tests on numaflow using TPS: 7000. Based on my continuous observation of the "isbsvc default js-x)" pod, the memory keeps rising until an Error (OOMKilled) occurs.
As shown in the figure:

This is my InterStepBufferService setting:

kind: InterStepBufferService
apiVersion: numaflow.numaproj.io/v1alpha1
metadata:
  name: default
  namespace: default
spec:
  jetstream:
    version: latest
    replicas: 3
    containerTemplate:
      resources:
        limits:
          cpu: "4"
          memory: 8Gi
    persistence:
      storageClassName: cbs
      volumeSize: 100Gi
    bufferConfig: |
      stream:
        # 0: Limits, 1: Interest, 2: WorkQueue
        retention: 0
        maxMsgs: 50000
        maxAge: 24h
        maxBytes: -1
        # 0: File, 1: Memory
        storage: 0         ### <- this switch to RAM only mode
        replicas: 3        ### <- this does not replicate data in the streams
        duplicates: 15m
  config:
    jetstream:
      streamConfig: |
        stream:
          duplicates: 15m
          maxage: 24h
          maxbytes: -1
          maxmsgs: 50000
          replicas: 3
          retention: 0
          storage: 0
  type: jetstream

This is my Pipeline setting:

kind: Pipeline
apiVersion: numaflow.numaproj.io/v1alpha1
metadata:
  name: bi-device-pipeline
  namespace: default
spec:
  vertices:
    - name: in
      source:
        kafka:
          brokers:
            - 10.0.2.224:9092
          topic: ad_pool_shiyuegame_com.device
          consumerGroup: huijuan-consumer-group-device
          tls: null
          config: >
            consumer:
              offsets:
                initial: -2 # -2 for sarama.OffsetOldest, -1 for sarama.OffsetNewest. Default to sarama.OffsetNewest.
          sasl: null
          kafkaVersion: 2.8.1
      containerTemplate:
        resources:
          limits:
            cpu: 300m
            memory: 322122547200m
          requests:
            cpu: 300m
            memory: 322122547200m
      imagePullSecrets:
        - name: fxzc-image
      scale:
        min: 1
        max: 1
      updateStrategy:
        type: RollingUpdate
        rollingUpdate:
          maxUnavailable: 25%
    - name: device
      udf:
        container:
          image: xx/device_mapper:latest
          resources: {}
        builtin: null
        groupBy: null
      containerTemplate:
        resources:
          limits:
            cpu: "1"
            memory: 2Gi
          requests:
            cpu: "1"
            memory: 2Gi
      imagePullSecrets:
        - name: fxzc-image
      limits:
        readBatchSize: 5000
        readTimeout: 30s
        bufferMaxLength: 20000
        bufferUsageLimit: 85
      scale:
        min: 1
        max: 1
      updateStrategy:
        type: RollingUpdate
        rollingUpdate:
          maxUnavailable: 25%
    - name: starrocks
      sink:
        udsink:
          container:
            image: xx/starrocks_sink:latest
            resources: {}
        retryStrategy: {}
      containerTemplate:
        resources:
          limits:
            cpu: "1"
            memory: 2Gi
          requests:
            cpu: "1"
            memory: 2Gi
      imagePullSecrets:
        - name: fxzc-image
      limits:
        readBatchSize: 5000
        readTimeout: 30s
        bufferMaxLength: 20000
        bufferUsageLimit: 85
      scale:
        min: 1
        max: 1
      updateStrategy:
        type: RollingUpdate
        rollingUpdate:
          maxUnavailable: 25%
  edges:
    - from: in
      to: device
      conditions: null
      onFull: retryUntilSuccess
    - from: device
      to: starrocks
      conditions: null
      onFull: retryUntilSuccess
  lifecycle:
    deleteGracePeriodSeconds: 30
    desiredPhase: Running
    pauseGracePeriodSeconds: 30
  limits:
    readBatchSize: 1000
    bufferMaxLength: 20000
    bufferUsageLimit: 85
    readTimeout: 30s
  watermark:
    maxDelay: 0s
  templates:
    daemon:
      imagePullSecrets:
        - name: fxzc-image
    job:
      imagePullSecrets:
        - name: fxzc-image
    sideInputsManager:
      imagePullSecrets:
        - name: fxzc-image
status:
  conditions:
    - type: Configured
      status: "True"
      lastTransitionTime: 2024-11-27T10:48:02Z
      reason: Successful
      message: Successful
    - type: DaemonServiceHealthy
      status: "True"
      lastTransitionTime: 2024-11-27T10:48:02Z
      reason: Successful
      message: Successful
    - type: Deployed
      status: "True"
      lastTransitionTime: 2024-11-27T10:48:02Z
      reason: Successful
      message: Successful
    - type: SideInputsManagersHealthy
      status: "True"
      lastTransitionTime: 2024-11-27T10:48:02Z
      reason: NoSideInputs
      message: No Side Inputs attached to the pipeline
    - type: VerticesHealthy
      status: "True"
      lastTransitionTime: 2024-11-27T10:48:02Z
      reason: Successful
      message: All vertices are healthy
  phase: Running
  lastUpdated: 2024-11-27T10:48:02Z
  vertexCount: 3
  sourceCount: 1
  sinkCount: 1
  udfCount: 1
  mapUDFCount: 1
  reduceUDFCount: 0
  observedGeneration: 4

numaflow:v1.4.0
k8s:v1.24.4
nats:2.10.17

[kohlisid]

Hello @qianbeibuzui
What is the msg size of the payloads that you are using?

[qianbeibuzui]

Hello @qianbeibuzui What is the msg size of the payloads that you are using?

Is it the 'max_mayload' setting item? No settings, using default values.

[kohlisid]

@qianbeibuzui I'm referring the to data that is getting stored in the stream. What is the average size of the messages flowing in the pipeline?

[kohlisid]

And you are using a limits queue, with file storage mode with Replication 3 correct?

[qianbeibuzui]

When concurrency is low, it is relatively stable, and it seems that the problem only occurs when concurrency is high.

[kohlisid]

Which concurrency are you referring to here specifically? Messages/sec?

[qianbeibuzui]

And you are using a limits queue, with file storage mode with Replication 3 correct?

Based on your previous advice, I have already used file storage and Replication=3

[qianbeibuzui]

Messages/sec

Messages/sec=7000/sec will reproduce

[qianbeibuzui]

@kohlisid Do you need me to provide any additional information to assist you in troubleshooting?

[kohlisid]

In my experiments with a single stream, I have seen the Jetstream server use higher memory than the stream size per se.

Would you like to try with a higher memory request and limit to see how much memory is getting consumed at a stable state for 7k?
Additionally, we can set a lower GOMEMLIMIT value to try and invoke the GC earlier

Would you like to try that out?

spec:
  jetstream:
    replicas: 3
    metricsContainerTemplate:
      env:
        - name: GOMEMLIMIT
          value: "19327352832"  
    reloaderContainerTemplate:
      env:
        - name: GOMEMLIMIT
          value: "19327352832"
    containerTemplate:
      env:
        - name: GOMEMLIMIT
          value: "19327352832"
      resources:
        limits:
          cpu: 6
          memory: 24Gi
        requests:
          cpu: 6
          memory: 24Gi

We have been tracking this with Jetstream as well
Ref:
nats-io/nats-server#5739
nats-io/nats-server#6163
nats-io/nats-server#6010

[qianbeibuzui]

In my experiments with a single stream, I have seen the Jetstream server use higher memory than the stream size per se.

Would you like to try with a higher memory request and limit to see how much memory is getting consumed at a stable state for 7k? Additionally, we can set a lower GOMEMLIMIT value to try and invoke the GC earlier

Would you like to try that out?

spec:
  jetstream:
    replicas: 3
    metricsContainerTemplate:
      env:
        - name: GOMEMLIMIT
          value: "19327352832"  
    reloaderContainerTemplate:
      env:
        - name: GOMEMLIMIT
          value: "19327352832"
    containerTemplate:
      env:
        - name: GOMEMLIMIT
          value: "19327352832"
      resources:
        limits:
          cpu: 6
          memory: 24Gi
        requests:
          cpu: 6
          memory: 24Gi

We have been tracking this with Jetstream as well Ref: nats-io/nats-server#5739 nats-io/nats-server#6163 nats-io/nats-server#6010

好的，我马上尝试一下