[runtime] Remove old redundant classes

  - StringRecord -> StringValue
  - IntegerRecord -> IntValue
  - FileRecord -> obsolete

flink-core/src/main/java/org/apache/flink/core/fs/Path.java

@@ -28,7 +28,6 @@
 import java.net.URISyntaxException;
 
 import org.apache.flink.core.io.IOReadableWritable;
-import org.apache.flink.core.io.StringRecord;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
 import org.apache.flink.util.OperatingSystem;
@@ -473,20 +472,19 @@ public void read(DataInputView in) throws IOException {
 
  final boolean isNotNull = in.readBoolean();
  if (isNotNull) {
- final String scheme = StringRecord.readString(in);
- final String userInfo = StringRecord.readString(in);
- final String host = StringRecord.readString(in);
+ final String scheme = StringUtils.readNullableString(in);
+ final String userInfo = StringUtils.readNullableString(in);
+ final String host = StringUtils.readNullableString(in);
  final int port = in.readInt();
- final String path = StringRecord.readString(in);
- final String query = StringRecord.readString(in);
- final String fragment = StringRecord.readString(in);
+ final String path = StringUtils.readNullableString(in);
+ final String query = StringUtils.readNullableString(in);
+ final String fragment = StringUtils.readNullableString(in);
 
  try {
  uri = new URI(scheme, userInfo, host, port, path, query, fragment);
  } catch (URISyntaxException e) {
- throw new IOException("Error reconstructing URI: " + StringUtils.stringifyException(e));
+ throw new IOException("Error reconstructing URI", e);
  }
-
  }
  }
 
@@ -498,13 +496,13 @@ public void write(DataOutputView out) throws IOException {
  out.writeBoolean(false);
  } else {
  out.writeBoolean(true);
- StringRecord.writeString(out, uri.getScheme());
- StringRecord.writeString(out, uri.getUserInfo());
- StringRecord.writeString(out, uri.getHost());
+ StringUtils.writeNullableString(uri.getScheme(), out);
+ StringUtils.writeNullableString(uri.getUserInfo(), out);
+ StringUtils.writeNullableString(uri.getHost(), out);
  out.writeInt(uri.getPort());
- StringRecord.writeString(out, uri.getPath());
- StringRecord.writeString(out, uri.getQuery());
- StringRecord.writeString(out, uri.getFragment());
+ StringUtils.writeNullableString(uri.getPath(), out);
+ StringUtils.writeNullableString(uri.getQuery(), out);
+ StringUtils.writeNullableString(uri.getFragment(), out);
  }
 
  }

flink-core/src/main/java/org/apache/flink/core/io/LocatableInputSplit.java

@@ -23,6 +23,7 @@
 
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.util.StringUtils;
 
 /**
  * A locatable input split is an input split referring to input data which is located on one or more hosts.
@@ -90,8 +91,8 @@ public String[] getHostnames() {
  public void write(DataOutputView out) throws IOException {
  out.writeInt(this.splitNumber);
  out.writeInt(this.hostnames.length);
- for (int i = 0; i < this.hostnames.length; i++) {
- StringRecord.writeString(out, this.hostnames[i]);
+ for (String hostname : this.hostnames) {
+ StringUtils.writeNullableString(hostname, out);
  }
  }
 
@@ -105,7 +106,7 @@ public void read(DataInputView in) throws IOException {
  } else {
  this.hostnames = new String[numHosts];
  for (int i = 0; i < numHosts; i++) {
- this.hostnames[i] = StringRecord.readString(in);
+ this.hostnames[i] = StringUtils.readNullableString(in);
  }
  }
  }