How It Works

Input (Kafka) → Processors (Transform) → Output (Flexprice)

Time:  0s    0.5s    1s    1.5s    2s
Event: 1─────2──────3─────4──────5
                                  └─▶ Batch sent (5 events, 2s elapsed)

pipeline:
  processors:
    - mapping: |
        root.properties = this.properties.map_each(p -> p.value.string())

pipeline:
  processors:
    - mapping: |
        # Only process successful requests
        root = if this.status >= 200 && this.status < 300 {
          this
        } else {
          deleted()
        }

input:
  kafka:
    fetch_buffer_cap: 256        # Fetch more messages at once
    checkpoint_limit: 1024       # Buffer messages in flight
    commit_period: 1s

output:
  flexprice:
    max_in_flight: 10  # Parallel API requests
    batching:
      count: 50
      period: 5s

output:
  flexprice:
    retry_max_attempts: 3
    retry_backoff:
      initial_interval: 1s
      max_interval: 60s

logger:
  level: INFO           # DEBUG, INFO, WARN, ERROR
  format: logfmt        # or json
  add_timestamp: true