如何处理CoderException:无法使用scio对空字符串进行编码

import org.apache.beam.sdk.transforms.Filter;
import org.apache.beam.sdk.values.PCollection;

PCollection<String> input = ...; // 输入数据
PCollection<String> filtered = input.apply(Filter.by((String s) -> !s.isEmpty()));

import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.StringUtf8Coder;

public class CustomStringCoder extends StringUtf8Coder {
    @Override
    public void encode(String value, OutputStream outStream) throws IOException {
        if (value == null || value.isEmpty()) {
            outStream.write(new byte[0]); // 编码空字符串为空字节
        } else {
            super.encode(value, outStream);
        }
    }

    @Override
    public String decode(InputStream inStream) throws IOException {
        byte[] bytes = new byte[inStream.available()];
        inStream.read(bytes);
        return new String(bytes, StandardCharsets.UTF_8);
    }
}

import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.values.PCollection;

PCollection<String> data = p.apply(Create.of("example", "").withCoder(new CustomStringCoder()));