kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix DataCarrier's `org.apache.skywalking.apm.commons.datacarrier.buffer.Buffer` implementation isn't activated in `IF_POSSIBLE` mode. (#6048)

This commit is contained in:
吴晟 Wu Sheng 2020-12-22 07:54:15 +08:00 committed by GitHub
parent 542691011c
commit e6265ba5be
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 15 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.md
View File

@ -19,6 +19,7 @@ Release Notes.
* Update `byte-buddy` to 1.10.19.
* Fix thrift plugin trace link broken when intermediate service does not mount agent
* Fix thrift plugin collects wrong args when the method without parameter.
* Fix DataCarrier's `org.apache.skywalking.apm.commons.datacarrier.buffer.Buffer` implementation isn't activated in `IF_POSSIBLE` mode.
#### OAP-Backend
* Make meter receiver support MAL.

18
apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/DataCarrier.java
View File

@ -44,10 +44,18 @@ public class DataCarrier<T> {
}
public DataCarrier(String name, String envPrefix, int channelSize, int bufferSize) {
this(name, envPrefix, channelSize, bufferSize, BufferStrategy.BLOCKING);
}
public DataCarrier(String name, String envPrefix, int channelSize, int bufferSize, BufferStrategy strategy) {
this.name = name;
bufferSize = EnvUtil.getInt(envPrefix + "_BUFFER_SIZE", bufferSize);
channelSize = EnvUtil.getInt(envPrefix + "_CHANNEL_SIZE", channelSize);
channels = new Channels<>(channelSize, bufferSize, new SimpleRollingPartitioner<T>(), BufferStrategy.BLOCKING);
channels = new Channels<>(channelSize, bufferSize, new SimpleRollingPartitioner<T>(), strategy);
}
public DataCarrier(int channelSize, int bufferSize, BufferStrategy strategy) {
this("DEFAULT", "DEFAULT", channelSize, bufferSize, strategy);
}
/**
@ -62,14 +70,6 @@ public class DataCarrier<T> {
return this;
}
/**
* override the strategy at runtime. Notice, {@link Channels} will override several channels one by one.
*/
public DataCarrier setBufferStrategy(BufferStrategy strategy) {
this.channels.setStrategy(strategy);
return this;
}
/**
* produce data to buffer, using the given {@link BufferStrategy}.
*

8
apm-commons/apm-datacarrier/src/test/java/org/apache/skywalking/apm/commons/datacarrier/DataCarrierTest.java
View File

@ -33,7 +33,7 @@ import org.powermock.api.support.membermodification.MemberModifier;
public class DataCarrierTest {
@Test
public void testCreateDataCarrier() throws IllegalAccessException {
DataCarrier<SampleData> carrier = new DataCarrier<>(5, 100);
DataCarrier<SampleData> carrier = new DataCarrier<>(5, 100, BufferStrategy.IF_POSSIBLE);
Channels<SampleData> channels = (Channels<SampleData>) (MemberModifier.field(DataCarrier.class, "channels")
.get(carrier));
@ -42,8 +42,7 @@ public class DataCarrierTest {
QueueBuffer<SampleData> buffer = channels.getBuffer(0);
Assert.assertEquals(100, buffer.getBufferSize());
Assert.assertEquals(MemberModifier.field(buffer.getClass(), "strategy").get(buffer), BufferStrategy.BLOCKING);
carrier.setBufferStrategy(BufferStrategy.IF_POSSIBLE);
Assert.assertEquals(MemberModifier.field(buffer.getClass(), "strategy").get(buffer), BufferStrategy.IF_POSSIBLE);
Assert.assertEquals(MemberModifier.field(buffer.getClass(), "strategy")
.get(buffer), BufferStrategy.IF_POSSIBLE);
@ -81,8 +80,7 @@ public class DataCarrierTest {
@Test
public void testIfPossibleProduce() throws IllegalAccessException {
DataCarrier<SampleData> carrier = new DataCarrier<SampleData>(2, 100);
carrier.setBufferStrategy(BufferStrategy.IF_POSSIBLE);
DataCarrier<SampleData> carrier = new DataCarrier<SampleData>(2, 100, BufferStrategy.IF_POSSIBLE);
for (int i = 0; i < 200; i++) {
Assert.assertTrue(carrier.produce(new SampleData().setName("d" + i)));

3
apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java
View File

@ -64,8 +64,7 @@ public class TraceSegmentServiceClient implements BootService, IConsumer<TraceSe
lastLogTime = System.currentTimeMillis();
segmentUplinkedCounter = 0;
segmentAbandonedCounter = 0;
carrier = new DataCarrier<>(CHANNEL_SIZE, BUFFER_SIZE);
carrier.setBufferStrategy(BufferStrategy.IF_POSSIBLE);
carrier = new DataCarrier<>(CHANNEL_SIZE, BUFFER_SIZE, BufferStrategy.IF_POSSIBLE);
carrier.consume(this, 1);
}

3
apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaTraceSegmentServiceClient.java
View File

@ -59,8 +59,7 @@ public class KafkaTraceSegmentServiceClient implements BootService, IConsumer<Tr
@Override
public void boot() {
carrier = new DataCarrier<>(CHANNEL_SIZE, BUFFER_SIZE);
carrier.setBufferStrategy(BufferStrategy.IF_POSSIBLE);
carrier = new DataCarrier<>(CHANNEL_SIZE, BUFFER_SIZE, BufferStrategy.IF_POSSIBLE);
carrier.consume(this, 1);
producer = ServiceManager.INSTANCE.findService(KafkaProducerManager.class).getProducer();