code: func (p *BulkProcessor) Start(ctx context.Context) error {
  	p.startedMu.Lock()
  	defer p.startedMu.Unlock()
  
  	if p.started {
  		return nil
  	}
  
  	// We must have at least one worker.
  	if p.numWorkers < 1 {
  		p.numWorkers = 1
  	}
  
  	p.requestsC = make(chan BulkableRequest)
  	p.executionId = 0
  	p.stats = newBulkProcessorStats(p.numWorkers)
  	p.stopReconnC = make(chan struct{})
  
  	// Create and start up workers.
  	p.workers = make([]*bulkWorker, p.numWorkers)
  	for i := 0; i < p.numWorkers; i++ {
  		p.workerWg.Add(1)
  		p.workers[i] = newBulkWorker(p, i)
  		go p.workers[i].work(ctx)
  	}
  
  	// Start the ticker for flush (if enabled)
  	if int64(p.flushInterval) > 0 {
  		p.flusherStopC = make(chan struct{})
  		go p.flusher(p.flushInterval)
  	}
  
  	p.started = true
  
  	return nil
  }
docs: // Start starts the bulk processor. If the processor is already started,
  // nil is returned.