apache/beam · error

pubsubio.Write only accepts PCollections of

Error message

pubsubio.Write only accepts PCollections of %v and %v, received %v

What it means

pubsubio.Write only accepts PCollections whose element type is []byte or the internal PubSub message type. If the input PCollection has any other type after the automatic []byte wrapping step, the library panics because it cannot marshal the elements into Pub/Sub messages.

Solutions

1. Convert elements to []byte (e.g. json.Marshal or proto.Marshal) before pubsubio.Write.
2. Use the package's message wrapper type if you need attributes/ordering keys.
3. Add an assert on col.Type() in tests to catch type changes early.

Example fix

// before
pubsubio.Write(s, project, topic, structCol)
// after
serialized := beam.ParDo(s, func(m MyMsg) []byte { b, _ := json.Marshal(m); return b }, structCol)
pubsubio.Write(s, project, topic, serialized)

if got := col.Type().Type(); got != reflectx.ByteSlice && got != pubSubMessageT {
    return fmt.Errorf("pubsubio.Write needs []byte or PubSub message, got %v", got)
}

defer func() {
    if r := recover(); r != nil {
        if s, ok := r.(string); ok && strings.Contains(s, "pubsubio.Write only accepts") {
            // handle
        } else { panic(r) }
    }
}()

When it happens

Trigger: Passing a PCollection of arbitrary structs, strings, or custom types directly to pubsubio.Write without first converting to []byte or the PubSub message type.

Common situations: Writing results of a ParDo that emit structs; forgetting to serialize with json.Marshal/protobuf before Write; changing an upstream PCollection's type and not updating the write path.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• Exactly one of Topic or Subscription must be set in…
• Unexpected number type
• AfterProcessingTime trigger set without a delay or…
• At least one subtrigger required for composite triggers.
• attempted to add namespace to missing coder id

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

Appendix: source

Thrown at sdks/go/pkg/beam/io/pubsubio/pubsubio.go:146

// Write writes PubSubMessages or []bytes to the given pubsub topic.
// Panics if the input pcollection type is not one of those two types.
//
// When given []bytes, they are first wrapped in PubSubMessages.
//
// Note: Doesn't function in batch pipelines.
func Write(s beam.Scope, project, topic string, col beam.PCollection) {
	s = s.Scope("pubsubio.Write")

	payload := &pipepb.PubSubWritePayload{
		Topic: pubsubx.MakeQualifiedTopicName(project, topic),
	}

	out := col
	if col.Type().Type() == reflectx.ByteSlice {
		out = beam.ParDo(s, wrapInMessage, col)
	}
	if out.Type().Type() != pubSubMessageT {
		panic(fmt.Sprintf("pubsubio.Write only accepts PCollections of %v and %v, received %v", pubSubMessageT, reflectx.ByteSlice, col.Type().Type()))
	}
	marshaled := beam.ParDo(s, marshalMessageFn, out)
	beam.External(s, writeURN, protox.MustEncode(payload), []beam.PCollection{marshaled}, nil, false)
}

View on GitHub (pinned to 12126d8942)