apache/beam · error

instruction %v no longer processing

Error message

instruction %v no longer processing

What it means

ScopedDataManager.open refuses to open a new data channel port when the manager for this instruction has already been closed. The Beam Go harness closes the ScopedDataManager once an instruction (bundle or process-bundle request) finishes processing; any late attempt to open a port afterwards is invalid state. This prevents writes to a data plane connection that no longer has a serving instruction.

Source

Thrown at sdks/go/pkg/beam/core/runtime/harness/datamgr.go:86

		return nil, err
	}
	return ch.OpenElementChan(ctx, id.PtransformID, s.instID, expectedTimerTransforms)
}

// OpenTimerWrite opens an io.WriteCloser on the given stream to write timers
func (s *ScopedDataManager) OpenTimerWrite(ctx context.Context, id exec.StreamID, family string) (io.WriteCloser, error) {
	ch, err := s.open(ctx, id.Port)
	if err != nil {
		return nil, err
	}
	return ch.OpenTimerWrite(ctx, id.PtransformID, s.instID, family), nil
}

func (s *ScopedDataManager) open(ctx context.Context, port exec.Port) (*DataChannel, error) {
	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		return nil, errors.Errorf("instruction %v no longer processing", s.instID)
	}
	s.openPorts = append(s.openPorts, port)
	local := s.mgr
	s.mu.Unlock()

	return local.Open(ctx, port) // don't hold lock over potentially slow operation
}

// Close prevents new IO for this instruction.
func (s *ScopedDataManager) Close() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.closed = true
	err := s.mgr.closeInstruction(s.instID, s.openPorts)
	s.mgr = nil
	return err
}

View on GitHub (pinned to 12126d8942)

Solutions

  1. Ensure all channel opens (OpenWrite/OpenElementChan/OpenTimerWrite) happen before the instruction's processing completes; do not retain writers beyond bundle lifetime
  2. Check that the instructionID used for the ScopedDataManager matches the currently processing instruction; do not reuse a closed manager
  3. In concurrent code, serialize opens with Close so no open lands after shutdown
  4. If this appears during pipeline shutdown, verify the runner is not sending extra bundle requests with a completed instruction ID

Example fix

// before
w := mgr.OpenWrite(ctx, port) // may fail if instruction already closed
// after
if err := mgr.EnsureOpen(ctx, instructionID); err != nil {
  return fmt.Errorf("instruction %s no longer processing; cannot open port: %w", instructionID, err)
}
w := mgr.OpenWrite(ctx, port)
Defensive patterns

Strategy: try-catch

Validate before calling

if mgr == nil || mgr.IsClosed() {
  return fmt.Errorf("scoped datamanager for %s already closed; refusing to open port", instID)
}

Type guard

func openable(m *ScopedDataManager) bool {
  m.mu.Lock(); defer m.mu.Unlock()
  return !m.closed
}

Try / catch

ch, err := mgr.OpenWrite(ctx, port)
if err != nil && strings.Contains(err.Error(), "no longer processing") {
  // bundle finished; drop late writer or re-open with a fresh instruction
  return ErrInstructionClosed
}

Prevention

When it happens

Trigger: Calling OpenWrite, OpenElementChan, or OpenTimerWrite after ScopedDataManager.Close() was invoked for the instruction ID; a runner issuing additional port opens after the bundle completed; concurrent open racing a close between checking and opening.

Common situations: Bundle already finished but the SDK still holds a DoFn that lazily opens a side-output or timer writer; harness shutdown ordering issues where logging/emitter goroutines outlive the instruction; retry logic that re-opens channels without recreating the ScopedDataManager.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/6824d4a8724b0040. Report an issue: GitHub.