apache/beam · error · RuntimeException

Not implemented

Error message

Not implemented ${fieldDescriptor.getMessageType().getFullName()}

What it means

When converting BYTES-typed TableRow fields whose value is a protobuf message, only recognized bytes-wrapper descriptor names (e.g. BytesValue) are handled; anything else hits this 'Not implemented' RuntimeException with the full message type name.

Solutions

Wrap the bytes in google.protobuf.BytesValue or set a raw byte[]/ByteArray in the TableRow
Unwrap the custom message to a ByteString and base64-encode it into the TableRow yourself
Verify the field's full name against BYTES_VALUE_DESCRIPTOR_NAMES in TableRowToStorageApiProto
Upgrade Beam for possibly added support
Add the descriptor name to BYTES_VALUE_DESCRIPTOR_NAMES if appropriate

Example fix

// before
row.set("payload", MyBlobProto.getDefaultInstance());
// after
row.set("payload", BytesValue.of(ByteString.copyFrom(bytes)));

Defensive patterns

Strategy: validation

Validate before calling

if (fieldValue instanceof Message && !BYTES_VALUE_DESCRIPTOR_NAMES.contains(((Message) fieldValue).getDescriptorForType().getName())) {
  throw new IllegalArgumentException("Unsupported bytes wrapper: " + ((Message) fieldValue).getDescriptorForType().getFullName());
}

Type guard

boolean isSupportedBytesWrapper(Object v) {
  return !(v instanceof Message)
      || BYTES_VALUE_DESCRIPTOR_NAMES.contains(((Message) v).getDescriptorForType().getName());
}

Try / catch

try {
  proto = TableRowToStorageApiProto.messageToValue(schema, row, options);
} catch (RuntimeException e) {
  if (e.getMessage().startsWith("Not implemented")) { /* DLQ or legacy path */ } else throw e;
}

Prevention

Store byte[] or BytesValue in BYTES fields, never custom messages
Unwrap to ByteString and base64-encode when building rows from protos
Check BYTES_VALUE_DESCRIPTOR_NAMES after Beam/protobuf upgrades

When it happens

Trigger: A TableRow field of BYTES schema holds a protobuf message that is not a recognized BytesValue wrapper when TableRowToStorageApiProto.messageToValue processes it.

Common situations: Custom byte-wrapper messages generated from user protos, proto-derived rows fed to BigQueryIO with the Storage API method, or mismatched wrapper types after a proto regeneration.

Related errors

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

Appendix: source

Thrown at sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowToStorageApiProto.java:1982

          default:
            return fieldValue.toString();
        }
      case BYTES:
        switch (fieldDescriptor.getType()) {
          case BYTES:
            return BaseEncoding.base64().encode(((ByteString) fieldValue).toByteArray());
          case STRING:
            return BaseEncoding.base64()
                .encode(((String) fieldValue).getBytes(StandardCharsets.UTF_8));
          case MESSAGE:
            Message message = (Message) fieldValue;
            if (BYTES_VALUE_DESCRIPTOR_NAMES.contains(fieldDescriptor.getMessageType().getName())) {
              ByteString byteString =
                  (ByteString)
                      message.getField(message.getDescriptorForType().findFieldByName("value"));
              return BaseEncoding.base64().encode(byteString.toByteArray());
            }
            throw new RuntimeException(
                "Not implemented " + fieldDescriptor.getMessageType().getFullName());
          default:
            return fieldValue.toString();
        }
      case TIMESTAMP:
        if (isProtoFieldTypeInteger(fieldDescriptor.getType())) {
          long epochMicros = Long.valueOf(fieldValue.toString());
          long epochSeconds = epochMicros / 1_000_000L;
          long nanoAdjustment = (epochMicros % 1_000_000L) * 1_000L;
          Instant instant = Instant.ofEpochSecond(epochSeconds, nanoAdjustment);
          return LocalDateTime.ofInstant(instant, ZoneOffset.UTC).format(TIMESTAMP_FORMATTER);
        } else if (fieldDescriptor.getType().equals(FieldDescriptor.Type.MESSAGE)) {
          Message message = (Message) fieldValue;
          String messageName = fieldDescriptor.getMessageType().getName();
          if (TIMESTAMP_VALUE_DESCRIPTOR_NAMES.contains(
              fieldDescriptor.getMessageType().getName())) {
            Descriptor descriptor = message.getDescriptorForType();
            long seconds = (long) message.getField(descriptor.findFieldByName("seconds"));

View on GitHub (pinned to 12126d8942)