@@ -0,0 +1,1206 @@
+package eastcircle.terasort;%0A%0Aimport java.io.IOException;%0A%0Aimport org.apache.flink.core.memory.DataInputView;%0Aimport org.apache.flink.core.memory.DataOutputView;%0Aimport org.apache.flink.core.memory.MemorySegment;%0Aimport org.apache.flink.types.NormalizableKey;%0Aimport org.apache.hadoop.io.Text;%0A%0Apublic final class OptimizedText implements NormalizableKey%3COptimizedText%3E %7B%0A%0A    private final Text text;%0A%0A    public OptimizedText () %7B%0A        this.text = new Text();%0A    %7D%0A%0A    public OptimizedText (Text from) %7B%0A        this.text = from;%0A    %7D%0A%0A    public Text getText() %7B%0A        return text;%0A    %7D%0A%0A    @Override%0A    public int getMaxNormalizedKeyLen() %7B%0A        return 10;%0A    %7D%0A%0A    @Override%0A    public void copyNormalizedKey(MemorySegment memory, int offset, int len) %7B%0A        memory.put(offset, text.getBytes(), 0, Math.min(text.getLength(), Math.min(10, len)));%0A    %7D%0A%0A    @Override%0A    public void write(DataOutputView out) throws IOException %7B%0A        text.write(out);%0A    %7D%0A%0A    @Override%0A    public void read(DataInputView in) throws IOException %7B%0A        text.readFields(in);%0A    %7D%0A%0A    @Override%0A    public int compareTo(OptimizedText o) %7B%0A        return this.text.compareTo(o.text);%0A    %7D%0A%7D%0A

modify FlinkTeraSort and add OptimizedText