diff --git a/oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/BufferFileUtils.java b/oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/BufferFileUtils.java index d39709bdc6..e27d688690 100644 --- a/oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/BufferFileUtils.java +++ b/oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/BufferFileUtils.java @@ -21,6 +21,9 @@ package org.apache.skywalking.oap.server.library.buffer; import java.util.Arrays; /** + * This class is a util for sort or build file name for the gRPC streaming data. + * Sort the files by the created time in order to read the data file sequential. + * * @author peng-yongsheng */ class BufferFileUtils { diff --git a/oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/DataStreamReader.java b/oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/DataStreamReader.java index 79b2c8e444..b8cf17bf92 100644 --- a/oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/DataStreamReader.java +++ b/oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/DataStreamReader.java @@ -20,6 +20,7 @@ package org.apache.skywalking.oap.server.library.buffer; import com.google.protobuf.*; import java.io.*; +import java.util.Objects; import java.util.concurrent.*; import org.apache.commons.io.FileUtils; import org.apache.commons.io.filefilter.PrefixFileFilter; @@ -78,8 +79,12 @@ public class DataStreamReader { private void openInputStream(File readingFile) { try { this.readingFile = readingFile; + if (Objects.nonNull(inputStream)) { + inputStream.close(); + } + inputStream = new FileInputStream(readingFile); - } catch (FileNotFoundException e) { + } catch (IOException e) { logger.error(e.getMessage(), e); } } diff --git a/oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/DataStreamWriter.java b/oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/DataStreamWriter.java index cb1a0c6065..d4d0c87c53 100644 --- a/oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/DataStreamWriter.java +++ b/oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/DataStreamWriter.java @@ -88,6 +88,7 @@ class DataStreamWriter { writeOffset.setOffset(position); if (position >= (FileUtils.ONE_MB * dataFileMaxSize)) { File writingFile = createNewFile(); + outputStream.close(); outputStream = FileUtils.openOutputStream(writingFile, true); } } catch (IOException e) { diff --git a/oap-server/server-library/library-buffer/src/test/java/org/apache/skywalking/oap/server/library/buffer/BufferFileUtilsTestCase.java b/oap-server/server-library/library-buffer/src/test/java/org/apache/skywalking/oap/server/library/buffer/BufferFileUtilsTestCase.java new file mode 100644 index 0000000000..3eb69b11e2 --- /dev/null +++ b/oap-server/server-library/library-buffer/src/test/java/org/apache/skywalking/oap/server/library/buffer/BufferFileUtilsTestCase.java @@ -0,0 +1,47 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.library.buffer; + +import java.util.*; +import org.junit.*; + +/** + * @author peng-yongsheng + */ +public class BufferFileUtilsTestCase { + + @Test + public void testSort() { + List fileNames = new ArrayList<>(); + fileNames.add("data-1.sw"); + fileNames.add("data-3.sw"); + fileNames.add("data-2.sw"); + fileNames.add("data-8.sw"); + fileNames.add("data-5.sw"); + + String[] files = fileNames.toArray(new String[0]); + BufferFileUtils.sort(files); + + Assert.assertEquals("data-1.sw", files[0]); + Assert.assertEquals("data-2.sw", files[1]); + Assert.assertEquals("data-3.sw", files[2]); + Assert.assertEquals("data-5.sw", files[3]); + Assert.assertEquals("data-8.sw", files[4]); + } +} diff --git a/oap-server/server-library/library-buffer/src/test/resources/log4j2.xml b/oap-server/server-library/library-buffer/src/test/resources/log4j2.xml index 6eb5b3fb98..41512643c4 100644 --- a/oap-server/server-library/library-buffer/src/test/resources/log4j2.xml +++ b/oap-server/server-library/library-buffer/src/test/resources/log4j2.xml @@ -20,7 +20,7 @@ - + diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/AgentDataMock.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/AgentDataMock.java index d2b7f03f32..924d1e205c 100644 --- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/AgentDataMock.java +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/AgentDataMock.java @@ -63,13 +63,15 @@ public class AgentDataMock { TimeUnit.SECONDS.sleep(10); - globalTraceId = UniqueIdBuilder.INSTANCE.create(); - serviceASegmentId = UniqueIdBuilder.INSTANCE.create(); - serviceBSegmentId = UniqueIdBuilder.INSTANCE.create(); - serviceCSegmentId = UniqueIdBuilder.INSTANCE.create(); - serviceAMock.mock(streamObserver, globalTraceId, serviceASegmentId, startTimestamp, false); - serviceBMock.mock(streamObserver, globalTraceId, serviceBSegmentId, serviceASegmentId, startTimestamp, false); - serviceCMock.mock(streamObserver, globalTraceId, serviceCSegmentId, serviceBSegmentId, startTimestamp, false); + for (int i = 0; i < 500; i++) { + globalTraceId = UniqueIdBuilder.INSTANCE.create(); + serviceASegmentId = UniqueIdBuilder.INSTANCE.create(); + serviceBSegmentId = UniqueIdBuilder.INSTANCE.create(); + serviceCSegmentId = UniqueIdBuilder.INSTANCE.create(); + serviceAMock.mock(streamObserver, globalTraceId, serviceASegmentId, startTimestamp, false); + serviceBMock.mock(streamObserver, globalTraceId, serviceBSegmentId, serviceASegmentId, startTimestamp, false); + serviceCMock.mock(streamObserver, globalTraceId, serviceCSegmentId, serviceBSegmentId, startTimestamp, false); + } streamObserver.onCompleted(); while (!IS_COMPLETED) {