@Override
	public void run(SourceContext<Row> ctx) throws Exception {
		final StringBuilder buffer = new StringBuilder();
		long attempt = 0;

		while (isRunning) {
			try (Socket socket = new Socket()) {
				currentSocket = socket;
				LOG.info("Connecting to server socket " + tableInfo.getHostname() + ':' + tableInfo.getPort());
				socket.connect(new InetSocketAddress(tableInfo.getHostname(), tableInfo.getPort()), CONNECTION_TIMEOUT_TIME);

				try (BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()))) {
					char[] cbuf = new char[8192];
					int bytesRead;
					while (isRunning && (bytesRead = reader.read(cbuf)) != -1) {
						buffer.append(cbuf, 0, bytesRead);
						int delimPos;
						String delimiter = tableInfo.getDelimiter();
						while (buffer.length() >= delimiter.length() && (delimPos = buffer.indexOf(delimiter)) != -1) {
							String record = buffer.substring(0, delimPos);
							// truncate trailing carriage return
							if ("\n".equals(delimiter) && "\r".endsWith(record)) {
								record = record.substring(0, record.length() - 1);
							}
							try {
								Row row = deserializationMetricWrapper.deserialize(record.getBytes());
								ctx.collect(row);
							} catch (Exception e) {
								LOG.error("parseData error ", e);
							} finally {
								buffer.delete(0, delimPos + delimiter.length());
							}
						}
					}
				}
			}

			// if we dropped out of this loop due to an EOF, sleep and retry
			if (isRunning) {
				attempt++;
				if (tableInfo.getMaxNumRetries() == -1 || attempt < tableInfo.getMaxNumRetries()) {
					Thread.sleep(DEFAULT_CONNECTION_RETRY_SLEEP);
				} else {
					// this should probably be here, but some examples expect simple exists of the stream source
					// throw new EOFException("Reached end of stream and reconnects are not enabled.");
					break;
				}
			}
		}

		// collect trailing data
		if (buffer.length() > 0) {
			try {
				Row row = deserializationMetricWrapper.deserialize(buffer.toString().getBytes());
				ctx.collect(row);
			} catch (Exception e) {
				LOG.error("parseData error ", e);
			}
		}
	}
Default connection timeout when connecting to the server socket (infinite).
serversocket/serversocket-source/src/main/java/com/dtstack/flink/sql/source/serversocket/CustomerSocketTextStreamFunction.java