项目中我要做的是，读取一个bin文件，根据指定协议拆解成数个数组。以下是拆解部分的代码。
public static void TboxPacketSpliter(FileInputStream inputStream, SourceContext ctx, String vvin, String messageType) throws IOException {
List<byte[]> byteList = new ArrayList<>();
byte[] readBuffer = new byte[16];
boolean ifHead = true;
//第一条数据单独处理一次
byte[] supplementBytes = new byte[]{(byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x00};
byte[] timestampBytes = new byte[8];
inputStream.read(timestampBytes);
byteList.add(supplementBytes);
byteList.add(timestampBytes);
while ((inputStream.read(readBuffer) != -1)) {
for (int i = 0; i < 8; i++) {
if (readBuffer[i] != (byte) 0x00) {
ifHead = false;
}
}
if (ifHead) {
byte[] outPutBytes = BytesCombine(byteList);
try {
ctx.collect(new Tuple3<>(outPutBytes, messageType, vvin));
} catch (Exception e) {
e.printStackTrace();
}
byteList.clear();
byteList.add(readBuffer);
} else {
byteList.add(readBuffer.clone());
ifHead = true;
}
}
inputStream.close();
}
org.apache.flink.util.FlinkRuntimeException: The Source Context has been closed already.
at org.apache.flink.streaming.api.operators.StreamSourceContexts$ClosedContext.throwException(StreamSourceContexts.java:172)
at org.apache.flink.streaming.api.operators.StreamSourceContexts$ClosedContext.collect(StreamSourceContexts.java:143)
at org.apache.flink.streaming.api.operators.StreamSourceContexts$SwitchingOnClose.collect(StreamSourceContexts.java:103)
at com.faw.autopilot.tool_test.TboxTools.TboxPacketSpliter(TboxTools.java:37)
at com.faw.autopilot.flink_main_test.FlinkSinkTest$SourceFromFile.run(FlinkSinkTest.java:104)
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:116)
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:73)
at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:323)
刚开始运行是没有问题的，当运行一分钟左右之后，对之后的每条数据都会爆出这个错误。问题是我并没有调用过close()方法，让我很迷惑。
希望各位能给予援助，万分感谢！
可能还是flink程序的问题，错误的字面含义貌似是flink停掉了，可以放到环境上的flink里边试试，排查flink的日志信息