natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'apache-3.1' into apache-3.2 

# Conflicts:
#	dubbo-dependencies-bom/pom.xml
#	dubbo-dependencies/dubbo-dependencies-zookeeper-curator5/pom.xml
#	dubbo-dependencies/dubbo-dependencies-zookeeper/pom.xml
#	pom.xml
This commit is contained in:
Albumen Kevin 2022-11-29 20:42:33 +08:00
parent b23cfa8c07 d07d89159f
commit 8c61f9b225
29 changed files with 491 additions and 267 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/registry/ZoneAwareClusterInvoker.java
View File

@ -46,7 +46,6 @@ import static org.apache.dubbo.common.constants.RegistryConstants.ZONE_KEY;
* 1. registry marked as 'preferred=true' has the highest priority.
* 2. check the zone the current request belongs, pick the registry that has the same zone first.
* 3. Evenly balance traffic between all registries based on each registry's weight.
* 4. Pick anyone that's available.
*/
public class ZoneAwareClusterInvoker<T> extends AbstractClusterInvoker<T> {
@ -112,8 +111,7 @@ public class ZoneAwareClusterInvoker<T> extends AbstractClusterInvoker<T> {
}
}
//if none available,just pick one
return invokers.get(0).invoke(invocation);
throw new RpcException("No provider available in " + invokers);
}
}

13
dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/registry/ZoneAwareClusterInvokerTest.java
View File

@ -20,6 +20,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.ClusterInvoker;
import org.apache.dubbo.rpc.cluster.Directory;
@ -168,6 +169,18 @@ class ZoneAwareClusterInvokerTest {
() -> zoneAwareClusterInvoker.invoke(invocation));
}
@Test
public void testNoAvailableInvoker() {
given(directory.getUrl()).willReturn(url);
given(directory.getConsumerUrl()).willReturn(url);
given(directory.list(invocation)).willReturn(new ArrayList<>(0));
zoneAwareClusterInvoker = new ZoneAwareClusterInvoker<>(directory);
Assertions.assertThrows(RpcException.class,
() -> zoneAwareClusterInvoker.invoke(invocation));
}
private ClusterInvoker newUnexpectedInvoker() {
return (ClusterInvoker) Proxy.newProxyInstance(getClass().getClassLoader(), new Class<?>[]{ClusterInvoker.class}, (proxy, method, args) -> {
if ("getUrl".equals(method.getName())) {

4
dubbo-common/src/main/java/org/apache/dubbo/common/constants/LoggerCodeConstants.java
View File

@ -71,9 +71,9 @@ public interface LoggerCodeConstants {
String COMMON_FAILED_OVERRIDE_FIELD = "0-24";
String COMMON_FAILED_LOAD_MAPPING_CACHE = "0-24";
String COMMON_FAILED_LOAD_MAPPING_CACHE = "0-25";
String COMMON_METADATA_PROCESSOR = "0-25";
String COMMON_METADATA_PROCESSOR = "0-26";
// registry module
String REGISTRY_ADDRESS_INVALID = "1-1";

6
dubbo-common/src/main/java/org/apache/dubbo/common/threadlocal/InternalThreadLocalMap.java
View File

@ -34,8 +34,12 @@ public final class InternalThreadLocalMap {
static final Object UNSET = new Object();
/**
* should not be modified after initialization,
* do not set as final due to unit test
*/
// Reference: https://hg.openjdk.java.net/jdk8/jdk8/jdk/file/tip/src/share/classes/java/util/ArrayList.java#l229
private static final int ARRAY_LIST_CAPACITY_MAX_SIZE = Integer.MAX_VALUE - 8;
static int ARRAY_LIST_CAPACITY_MAX_SIZE = Integer.MAX_VALUE - 8;
private static final int ARRAY_LIST_CAPACITY_EXPAND_THRESHOLD = 1 << 30;

14
dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/support/AbortPolicyWithReport.java
View File

@ -60,7 +60,7 @@ public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy {
private final URL url;
private static volatile long lastPrintTime = 0;
protected static volatile long lastPrintTime = 0;
private static final long TEN_MINUTES_MILLS = 10 * 60 * 1000;
@ -68,7 +68,7 @@ public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy {
private static final String DEFAULT_DATETIME_FORMAT = "yyyy-MM-dd_HH:mm:ss";
private static Semaphore guard = new Semaphore(1);
protected static Semaphore guard = new Semaphore(1);
private static final String USER_HOME = System.getProperty("user.home");
@ -161,8 +161,8 @@ public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy {
//try-with-resources
try (FileOutputStream jStackStream = new FileOutputStream(
new File(dumpPath, "Dubbo_JStack.log" + "." + dateStr))) {
JVMUtil.jstack(jStackStream);
} catch (Throwable t) {
jstack(jStackStream);
} catch (Exception t) {
logger.error(COMMON_UNEXPECTED_CREATE_DUMP, "", "", "dump jStack error", t);
} finally {
guard.release();
@ -174,7 +174,11 @@ public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy {
}
private String getDumpPath() {
protected void jstack(FileOutputStream jStackStream) throws Exception {
JVMUtil.jstack(jStackStream);
}
protected String getDumpPath() {
final String dumpPath = url.getParameter(DUMP_DIRECTORY);
if (StringUtils.isEmpty(dumpPath)) {
return USER_HOME;

20
dubbo-common/src/test/java/org/apache/dubbo/common/concurrent/CompletableFutureTaskTest.java
View File

@ -18,6 +18,7 @@ package org.apache.dubbo.common.concurrent;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.concurrent.CompletableFuture;
@ -28,6 +29,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
@ -54,33 +56,33 @@ class CompletableFutureTaskTest {
@Test
void testRunnableResponse() throws ExecutionException, InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
CompletableFuture<Boolean> completableFuture = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(500);
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
throw new RuntimeException(e);
}
return true;
}, executor);
Assertions.assertNull(completableFuture.getNow(null));
latch.countDown();
Boolean result = completableFuture.get();
assertThat(result, is(true));
}
@Test
void testListener() throws InterruptedException {
AtomicBoolean run = new AtomicBoolean(false);
CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
run.set(true);
return "hello";
}, executor);
final CountDownLatch countDownLatch = new CountDownLatch(1);
completableFuture.thenRunAsync(countDownLatch::countDown);
countDownLatch.await();
Assertions.assertTrue(run.get());
}
@ -93,4 +95,4 @@ class CompletableFutureTaskTest {
completableFuture.thenRunAsync(mock(Runnable.class), mockedExecutor).whenComplete((s, e) ->
verify(mockedExecutor, times(1)).execute(any(Runnable.class)));
}
}
}

2
dubbo-common/src/test/java/org/apache/dubbo/common/config/configcenter/file/FileSystemDynamicConfigurationTest.java
View File

@ -183,4 +183,4 @@ class FileSystemDynamicConfigurationTest {
//
// assertEquals(new TreeSet(asList("A", "B", "C")), configuration.getConfigKeys(DEFAULT_GROUP));
// }
}
}

19
dubbo-common/src/test/java/org/apache/dubbo/common/threadlocal/InternalThreadLocalTest.java
View File

@ -28,6 +28,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport;
import static org.awaitility.Awaitility.await;
import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.CoreMatchers.not;
@ -66,7 +67,7 @@ class InternalThreadLocalTest {
t.start();
}
Thread.sleep(2000);
await().until(index::get, is(THREADS));
}
@Test
@ -232,18 +233,23 @@ class InternalThreadLocalTest {
@Test
void testConstructionWithIndex() throws Exception {
int ARRAY_LIST_CAPACITY_MAX_SIZE = Integer.MAX_VALUE - 8;
// reset ARRAY_LIST_CAPACITY_MAX_SIZE to speed up
int NEW_ARRAY_LIST_CAPACITY_MAX_SIZE = 8;
Field nextIndexField =
InternalThreadLocalMap.class.getDeclaredField("NEXT_INDEX");
nextIndexField.setAccessible(true);
AtomicInteger nextIndex = (AtomicInteger) nextIndexField.get(AtomicInteger.class);
int arrayListCapacityMaxSize = InternalThreadLocalMap.ARRAY_LIST_CAPACITY_MAX_SIZE;
int nextIndex_before = nextIndex.get();
nextIndex.set(0);
final AtomicReference<Throwable> throwable = new AtomicReference<Throwable>();
try {
while (nextIndex.get() < ARRAY_LIST_CAPACITY_MAX_SIZE) {
InternalThreadLocalMap.ARRAY_LIST_CAPACITY_MAX_SIZE = NEW_ARRAY_LIST_CAPACITY_MAX_SIZE;
while (nextIndex.get() < NEW_ARRAY_LIST_CAPACITY_MAX_SIZE) {
new InternalThreadLocal<Boolean>();
}
assertEquals(ARRAY_LIST_CAPACITY_MAX_SIZE - 1, InternalThreadLocalMap.lastVariableIndex());
assertEquals(NEW_ARRAY_LIST_CAPACITY_MAX_SIZE - 1, InternalThreadLocalMap.lastVariableIndex());
try {
new InternalThreadLocal<Boolean>();
} catch (Throwable t) {
@ -252,10 +258,11 @@ class InternalThreadLocalTest {
// Assert the max index cannot greater than (ARRAY_LIST_CAPACITY_MAX_SIZE - 1)
assertThat(throwable.get(), is(instanceOf(IllegalStateException.class)));
// Assert the index was reset to ARRAY_LIST_CAPACITY_MAX_SIZE after it reaches ARRAY_LIST_CAPACITY_MAX_SIZE
assertEquals(ARRAY_LIST_CAPACITY_MAX_SIZE - 1, InternalThreadLocalMap.lastVariableIndex());
assertEquals(NEW_ARRAY_LIST_CAPACITY_MAX_SIZE - 1, InternalThreadLocalMap.lastVariableIndex());
} finally {
// Restore the index
nextIndex.set(nextIndex_before);
InternalThreadLocalMap.ARRAY_LIST_CAPACITY_MAX_SIZE = arrayListCapacityMaxSize;
}
}
@ -279,4 +286,4 @@ class InternalThreadLocalTest {
// Assert the expanded size is not overflowed to negative value
assertThat(throwable.get(), is(not(instanceOf(NegativeArraySizeException.class))));
}
}
}

77
dubbo-common/src/test/java/org/apache/dubbo/common/threadpool/manager/ExecutorRepositoryTest.java
View File

@ -25,9 +25,11 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.awaitility.Awaitility.await;
class ExecutorRepositoryTest {
private ApplicationModel applicationModel;
@ -93,52 +95,37 @@ class ExecutorRepositoryTest {
@Test
void testSharedExecutor() throws Exception {
ExecutorService sharedExecutor = executorRepository.getSharedExecutor();
MockTask task1 = new MockTask(2000);
MockTask task2 = new MockTask(100);
MockTask task3 = new MockTask(200);
sharedExecutor.execute(task1);
sharedExecutor.execute(task2);
sharedExecutor.submit(task3);
Thread.sleep(150);
Assertions.assertTrue(task1.isRunning());
Assertions.assertFalse(task1.isDone());
Assertions.assertTrue(task2.isRunning());
Assertions.assertTrue(task2.isDone());
Assertions.assertTrue(task3.isRunning());
Assertions.assertFalse(task3.isDone());
Thread.sleep(200);
Assertions.assertTrue(task3.isDone());
Assertions.assertFalse(task1.isDone());
}
private static class MockTask implements Runnable {
private long waitTimeMS;
private AtomicBoolean running = new AtomicBoolean();
private AtomicBoolean done = new AtomicBoolean();
public MockTask(long waitTimeMS) {
this.waitTimeMS = waitTimeMS;
}
@Override
public void run() {
running.set(true);
CountDownLatch latch = new CountDownLatch(3);
CountDownLatch latch1 = new CountDownLatch(1);
sharedExecutor.execute(()->{
latch.countDown();
try {
Thread.sleep(waitTimeMS);
latch1.await();
} catch (InterruptedException e) {
e.printStackTrace();
throw new RuntimeException(e);
}
done.set(true);
}
});
sharedExecutor.execute(()->{
latch.countDown();
try {
latch1.await();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
sharedExecutor.submit(()->{
latch.countDown();
try {
latch1.await();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
public boolean isDone() {
return done.get();
}
public boolean isRunning() {
return running.get();
}
await().until(()->latch.getCount() == 0);
Assertions.assertEquals(3, ((ThreadPoolExecutor)sharedExecutor).getActiveCount());
latch1.countDown();
await().until(()->((ThreadPoolExecutor)sharedExecutor).getActiveCount() == 0);
Assertions.assertEquals(3, ((ThreadPoolExecutor)sharedExecutor).getCompletedTaskCount());
}
}
}

78
dubbo-common/src/test/java/org/apache/dubbo/common/threadpool/manager/FrameworkExecutorRepositoryTest.java
View File

@ -23,8 +23,11 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.ThreadPoolExecutor;
import static org.awaitility.Awaitility.await;
class FrameworkExecutorRepositoryTest {
private FrameworkModel frameworkModel;
@ -51,52 +54,37 @@ class FrameworkExecutorRepositoryTest {
@Test
void testSharedExecutor() throws Exception {
ExecutorService sharedExecutor = frameworkExecutorRepository.getSharedExecutor();
FrameworkExecutorRepositoryTest.MockTask task1 = new FrameworkExecutorRepositoryTest.MockTask(2000);
FrameworkExecutorRepositoryTest.MockTask task2 = new FrameworkExecutorRepositoryTest.MockTask(100);
FrameworkExecutorRepositoryTest.MockTask task3 = new FrameworkExecutorRepositoryTest.MockTask(200);
sharedExecutor.execute(task1);
sharedExecutor.execute(task2);
sharedExecutor.submit(task3);
Thread.sleep(150);
Assertions.assertTrue(task1.isRunning());
Assertions.assertFalse(task1.isDone());
Assertions.assertTrue(task2.isRunning());
Assertions.assertTrue(task2.isDone());
Assertions.assertTrue(task3.isRunning());
Assertions.assertFalse(task3.isDone());
Thread.sleep(200);
Assertions.assertTrue(task3.isDone());
Assertions.assertFalse(task1.isDone());
}
private static class MockTask implements Runnable {
private long waitTimeMS;
private AtomicBoolean running = new AtomicBoolean();
private AtomicBoolean done = new AtomicBoolean();
public MockTask(long waitTimeMS) {
this.waitTimeMS = waitTimeMS;
}
@Override
public void run() {
running.set(true);
CountDownLatch latch = new CountDownLatch(3);
CountDownLatch latch1 = new CountDownLatch(1);
sharedExecutor.execute(()->{
latch.countDown();
try {
Thread.sleep(waitTimeMS);
latch1.await();
} catch (InterruptedException e) {
e.printStackTrace();
throw new RuntimeException(e);
}
done.set(true);
}
});
sharedExecutor.execute(()->{
latch.countDown();
try {
latch1.await();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
sharedExecutor.submit(()->{
latch.countDown();
try {
latch1.await();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
public boolean isDone() {
return done.get();
}
public boolean isRunning() {
return running.get();
}
await().until(()->latch.getCount() == 0);
Assertions.assertEquals(3, ((ThreadPoolExecutor)sharedExecutor).getActiveCount());
latch1.countDown();
await().until(()->((ThreadPoolExecutor)sharedExecutor).getActiveCount() == 0);
Assertions.assertEquals(3, ((ThreadPoolExecutor)sharedExecutor).getCompletedTaskCount());
}
}
}

93
dubbo-common/src/test/java/org/apache/dubbo/common/threadpool/serial/SerializingExecutorTest.java
View File

@ -18,7 +18,7 @@
package org.apache.dubbo.common.threadpool.serial;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.util.HashMap;
@ -26,44 +26,95 @@ import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.awaitility.Awaitility.await;
class SerializingExecutorTest {
protected static SerializingExecutor serializingExecutor;
private ExecutorService service;
private SerializingExecutor serializingExecutor;
@BeforeAll
public static void before() {
ExecutorService service = Executors.newFixedThreadPool(4);
@BeforeEach
public void before() {
service = Executors.newFixedThreadPool(4);
serializingExecutor = new SerializingExecutor(service);
}
@Test
void test1() throws InterruptedException {
int n = 2;
int eachCount = 1000;
int total = n * eachCount;
int sleepMillis = 10;
void testSerial() throws InterruptedException {
int total = 10000;
Map<String, Integer> map = new HashMap<>();
map.put("val", 0);
CountDownLatch downLatch = new CountDownLatch(total);
Semaphore semaphore = new Semaphore(1);
CountDownLatch startLatch = new CountDownLatch(1);
AtomicBoolean failed = new AtomicBoolean(false);
for (int i = 0; i < total; i++) {
final int index = i;
Thread.sleep(ThreadLocalRandom.current().nextInt(sleepMillis));
serializingExecutor.execute(() -> {
if (!semaphore.tryAcquire()) {
System.out.println("Concurrency");
failed.set(true);
}
try {
Thread.sleep(ThreadLocalRandom.current().nextInt(sleepMillis));
startLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
throw new RuntimeException(e);
}
int num = map.get("val");
map.put("val", num + 1);
downLatch.countDown();
Assertions.assertEquals(num, index);
if (num != index) {
System.out.println("Index error. Excepted :" + index + " but actual: " + num);
failed.set(true);
}
semaphore.release();
});
}
downLatch.await(3, TimeUnit.SECONDS);
Assertions.assertEquals(total, map.get("val"));
startLatch.countDown();
await().until(() -> map.get("val") == total);
Assertions.assertFalse(failed.get());
}
}
@Test
void testNonSerial() {
int total = 10;
Map<String, Integer> map = new HashMap<>();
map.put("val", 0);
Semaphore semaphore = new Semaphore(1);
CountDownLatch startLatch = new CountDownLatch(1);
AtomicBoolean failed = new AtomicBoolean(false);
for (int i = 0; i < total; i++) {
final int index = i;
service.execute(() -> {
if (!semaphore.tryAcquire()) {
failed.set(true);
}
try {
startLatch.await();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
int num = map.get("val");
map.put("val", num + 1);
if (num != index) {
failed.set(true);
}
semaphore.release();
});
}
await().until(() -> ((ThreadPoolExecutor) service).getActiveCount() == 4);
startLatch.countDown();
await().until(() -> ((ThreadPoolExecutor) service).getCompletedTaskCount() == total);
Assertions.assertTrue(failed.get());
}
}

63
dubbo-common/src/test/java/org/apache/dubbo/common/threadpool/support/AbortPolicyWithReportTest.java
View File

@ -20,31 +20,42 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.threadpool.event.ThreadPoolExhaustedEvent;
import org.apache.dubbo.common.threadpool.event.ThreadPoolExhaustedListener;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.io.FileOutputStream;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicReference;
import static org.apache.dubbo.common.constants.CommonConstants.OS_NAME_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.OS_WIN_PREFIX;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
class AbortPolicyWithReportTest {
@Test
void jStackDumpTest() throws InterruptedException {
URL url = URL.valueOf("dubbo://admin:hello1234@10.20.130.230:20880/context/path?dump.directory=/tmp&version=1.0.0&application=morgan&noValue=");
AbortPolicyWithReport abortPolicyWithReport = new AbortPolicyWithReport("Test", url);
AtomicReference<FileOutputStream> fileOutputStream = new AtomicReference<>();
try {
abortPolicyWithReport.rejectedExecution(() -> System.out.println("hello"), (ThreadPoolExecutor) Executors.newFixedThreadPool(1));
} catch (RejectedExecutionException rj) {
// ignore
}
Thread.sleep(1000);
AbortPolicyWithReport abortPolicyWithReport = new AbortPolicyWithReport("Test", url) {
@Override
protected void jstack(FileOutputStream jStackStream) throws Exception {
fileOutputStream.set(jStackStream);
}
};
ExecutorService executorService = Executors.newFixedThreadPool(1);
AbortPolicyWithReport.lastPrintTime = 0;
Assertions.assertThrows(RejectedExecutionException.class, () -> {
abortPolicyWithReport.rejectedExecution(() -> System.out.println("hello"), (ThreadPoolExecutor) executorService);
});
await().until(() -> AbortPolicyWithReport.guard.availablePermits() == 1);
Assertions.assertNotNull(fileOutputStream.get());
}
@Test
@ -52,22 +63,11 @@ class AbortPolicyWithReportTest {
final String dumpDirectory = dumpDirectoryCannotBeCreated();
URL url = URL.valueOf("dubbo://admin:hello1234@10.20.130.230:20880/context/path?dump.directory="
+ dumpDirectory
+ "&version=1.0.0&application=morgan&noValue=true");
+ dumpDirectory
+ "&version=1.0.0&application=morgan&noValue=true");
AbortPolicyWithReport abortPolicyWithReport = new AbortPolicyWithReport("Test", url);
try {
abortPolicyWithReport.rejectedExecution(new Runnable() {
@Override
public void run() {
System.out.println("hello");
}
}, (ThreadPoolExecutor) Executors.newFixedThreadPool(1));
} catch (RejectedExecutionException rj) {
// ignore
}
Thread.sleep(1000);
Assertions.assertEquals(System.getProperty("user.home"), abortPolicyWithReport.getDumpPath());
}
private String dumpDirectoryCannotBeCreated() {
@ -85,22 +85,11 @@ class AbortPolicyWithReportTest {
final String dumpDirectory = UUID.randomUUID().toString();
URL url = URL.valueOf("dubbo://admin:hello1234@10.20.130.230:20880/context/path?dump.directory="
+ dumpDirectory
+ "&version=1.0.0&application=morgan&noValue=true");
+ dumpDirectory
+ "&version=1.0.0&application=morgan&noValue=true");
AbortPolicyWithReport abortPolicyWithReport = new AbortPolicyWithReport("Test", url);
try {
abortPolicyWithReport.rejectedExecution(new Runnable() {
@Override
public void run() {
System.out.println("hello");
}
}, (ThreadPoolExecutor) Executors.newFixedThreadPool(1));
} catch (RejectedExecutionException rj) {
// ignore
}
Thread.sleep(1000);
Assertions.assertNotEquals(System.getProperty("user.home"), abortPolicyWithReport.getDumpPath());
}
@Test
@ -128,4 +117,4 @@ class AbortPolicyWithReportTest {
return threadPoolExhaustedEvent;
}
}
}
}

161
dubbo-common/src/test/java/org/apache/dubbo/common/threadpool/support/eager/EagerThreadPoolExecutorTest.java
View File

@ -24,11 +24,18 @@ import org.apache.dubbo.common.url.component.ServiceConfigURL;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import static org.awaitility.Awaitility.await;
class EagerThreadPoolExecutorTest {
@ -55,6 +62,7 @@ class EagerThreadPoolExecutorTest {
* We can see , when the core threads are in busy,
* the thread pool create thread (but thread nums always less than max) instead of put task into queue.
*/
@Disabled("replaced to testEagerThreadPoolFast for performance")
@Test
void testEagerThreadPool() throws Exception {
String name = "eager-tf";
@ -67,19 +75,19 @@ class EagerThreadPoolExecutorTest {
//init queue and executor
TaskQueue<Runnable> taskQueue = new TaskQueue<Runnable>(queues);
final EagerThreadPoolExecutor executor = new EagerThreadPoolExecutor(cores,
threads,
alive,
TimeUnit.MILLISECONDS,
taskQueue,
new NamedThreadFactory(name, true),
new AbortPolicyWithReport(name, URL));
threads,
alive,
TimeUnit.MILLISECONDS,
taskQueue,
new NamedThreadFactory(name, true),
new AbortPolicyWithReport(name, URL));
taskQueue.setExecutor(executor);
for (int i = 0; i < 15; i++) {
Thread.sleep(50);
executor.execute(() -> {
System.out.println("thread number in current pool：" + executor.getPoolSize() + ", task number in task queue：" + executor.getQueue()
.size() + " executor size: " + executor.getPoolSize());
.size() + " executor size: " + executor.getPoolSize());
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
@ -92,17 +100,78 @@ class EagerThreadPoolExecutorTest {
Assertions.assertEquals(executor.getPoolSize(), cores, "more than cores threads alive!");
}
@Test
void testEagerThreadPoolFast() throws Exception {
String name = "eager-tf";
int queues = 5;
int cores = 5;
int threads = 10;
// alive 1 second
long alive = 1000;
//init queue and executor
TaskQueue<Runnable> taskQueue = new TaskQueue<Runnable>(queues);
final EagerThreadPoolExecutor executor = new EagerThreadPoolExecutor(cores,
threads,
alive,
TimeUnit.MILLISECONDS,
taskQueue,
new NamedThreadFactory(name, true),
new AbortPolicyWithReport(name, URL));
taskQueue.setExecutor(executor);
CountDownLatch countDownLatch1 = new CountDownLatch(1);
for (int i = 0; i < 10; i++) {
executor.execute(() -> {
try {
countDownLatch1.await();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
}
await().until(() -> executor.getPoolSize() == 10);
Assertions.assertEquals(10, executor.getActiveCount());
CountDownLatch countDownLatch2 = new CountDownLatch(1);
AtomicBoolean started = new AtomicBoolean(false);
for (int i = 0; i < 5; i++) {
executor.execute(() -> {
started.set(true);
try {
countDownLatch2.await();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
}
await().until(() -> executor.getQueue().size() == 5);
Assertions.assertEquals(10, executor.getActiveCount());
Assertions.assertEquals(10, executor.getPoolSize());
Assertions.assertFalse(started.get());
countDownLatch1.countDown();
await().until(() -> executor.getActiveCount() == 5);
Assertions.assertTrue(started.get());
countDownLatch2.countDown();
await().until(() -> executor.getActiveCount() == 0);
await().until(() -> executor.getPoolSize() == cores);
}
@Test
void testSPI() {
ExecutorService executorService = (ExecutorService) ExtensionLoader.getExtensionLoader(ThreadPool.class)
.getExtension("eager")
.getExecutor(URL);
.getExtension("eager")
.getExecutor(URL);
Assertions.assertEquals("EagerThreadPoolExecutor", executorService.getClass()
.getSimpleName(), "test spi fail!");
}
@Test
void testEagerThreadPool_rejectExecution() throws Exception {
void testEagerThreadPool_rejectExecution1() throws Exception {
String name = "eager-tf";
int cores = 1;
int threads = 3;
@ -119,20 +188,80 @@ class EagerThreadPoolExecutorTest {
new AbortPolicyWithReport(name, URL));
taskQueue.setExecutor(executor);
CountDownLatch countDownLatch = new CountDownLatch(1);
Runnable runnable = () -> {
System.out.println("thread number in current pool: " + executor.getPoolSize() + ", task number is task queue: " + executor.getQueue().size());
try {
Thread.sleep(1000);
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
throw new RuntimeException(e);
}
};
for (int i = 0; i < 5; i++) {
Thread.sleep(50);
executor.execute(runnable);
}
await().until(() -> executor.getPoolSize() == threads);
await().until(() -> executor.getQueue().size() == queues);
Assertions.assertThrows(RejectedExecutionException.class, () -> executor.execute(runnable));
Thread.sleep(10000);
countDownLatch.countDown();
await().until(() -> executor.getActiveCount() == 0);
executor.execute(runnable);
}
}
@Test
void testEagerThreadPool_rejectExecution2() throws Exception {
String name = "eager-tf";
int cores = 1;
int threads = 3;
int queues = 2;
long alive = 1000;
// init queue and executor
AtomicReference<Runnable> runnableWhenRetryOffer = new AtomicReference<>();
TaskQueue<Runnable> taskQueue = new TaskQueue<Runnable>(queues) {
@Override
public boolean retryOffer(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {
if (runnableWhenRetryOffer.get() != null) {
runnableWhenRetryOffer.get().run();
}
return super.retryOffer(o, timeout, unit);
}
};
final EagerThreadPoolExecutor executor = new EagerThreadPoolExecutor(cores,
threads,
alive, TimeUnit.MILLISECONDS,
taskQueue,
new NamedThreadFactory(name, true),
new AbortPolicyWithReport(name, URL));
taskQueue.setExecutor(executor);
Semaphore semaphore = new Semaphore(0);
Runnable runnable = () -> {
try {
semaphore.acquire();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
};
for (int i = 0; i < 5; i++) {
executor.execute(runnable);
}
await().until(() -> executor.getPoolSize() == threads);
await().until(() -> executor.getQueue().size() == queues);
Assertions.assertThrows(RejectedExecutionException.class, () -> executor.execute(runnable));
runnableWhenRetryOffer.set(() -> {
semaphore.release();
await().until(() -> executor.getCompletedTaskCount() == 1);
});
executor.execute(runnable);
semaphore.release(5);
await().until(() -> executor.getActiveCount() == 0);
}
}

23
dubbo-common/src/test/java/org/apache/dubbo/common/timer/HashedWheelTimerTest.java
View File

@ -28,6 +28,8 @@ import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import static org.awaitility.Awaitility.await;
class HashedWheelTimerTest {
private CountDownLatch tryStopTaskCountDownLatch = new CountDownLatch(1);
private CountDownLatch errorTaskCountDownLatch = new CountDownLatch(1);
@ -140,10 +142,11 @@ class HashedWheelTimerTest {
TimeUnit.MILLISECONDS,
8, 8);
EmptyTask emptyTask = new EmptyTask();
Assertions.assertThrows(RuntimeException.class,
() -> timer.newTimeout(null, 5, TimeUnit.SECONDS));
Assertions.assertThrows(RuntimeException.class,
() -> timer.newTimeout(new EmptyTask(), 5, null));
() -> timer.newTimeout(emptyTask, 5, null));
Timeout timeout = timer.newTimeout(new ErrorTask(), 10, TimeUnit.MILLISECONDS);
errorTaskCountDownLatch.await();
@ -152,27 +155,27 @@ class HashedWheelTimerTest {
Assertions.assertNotNull(timeout.toString());
Assertions.assertEquals(timeout.timer(), timer);
timeout = timer.newTimeout(new EmptyTask(), 1000, TimeUnit.SECONDS);
timeout = timer.newTimeout(emptyTask, 1000, TimeUnit.SECONDS);
timeout.cancel();
Assertions.assertTrue(timeout.isCancelled());
List<Timeout> timeouts = new LinkedList<>();
for (; timer.pendingTimeouts() < 8; ) {
BlockTask blockTask = new BlockTask();
while (timer.pendingTimeouts() < 8) {
// to trigger maxPendingTimeouts
timeout = timer.newTimeout(new BlockTask(), -1, TimeUnit.MILLISECONDS);
timeout = timer.newTimeout(blockTask, -1, TimeUnit.MILLISECONDS);
timeouts.add(timeout);
Assertions.assertNotNull(timeout.toString());
}
Assertions.assertEquals(timer.pendingTimeouts(), 8);
Assertions.assertEquals(8, timer.pendingTimeouts());
// this will throw an exception because of maxPendingTimeouts
Assertions.assertThrows(RuntimeException.class,
() -> timer.newTimeout(new BlockTask(), 1, TimeUnit.MILLISECONDS));
() -> timer.newTimeout(blockTask, 1, TimeUnit.MILLISECONDS));
timeout = timeouts.get(2);
Timeout secondTimeout = timeouts.get(2);
// wait until the task expired
Thread.sleep(100);
Assertions.assertTrue(timeout.isExpired());
await().until(secondTimeout::isExpired);
timer.stop();
}
@ -195,4 +198,4 @@ class HashedWheelTimerTest {
() -> timer.newTimeout(new EmptyTask(), 5, TimeUnit.SECONDS));
}
}
}

13
dubbo-common/src/test/java/org/apache/dubbo/common/utils/ExecutorUtilTest.java
View File

@ -28,10 +28,11 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import static org.apache.dubbo.common.constants.CommonConstants.THREAD_NAME_KEY;
import static org.awaitility.Awaitility.await;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ -62,9 +63,11 @@ class ExecutorUtilTest {
when(executor.awaitTermination(20, TimeUnit.MILLISECONDS)).thenReturn(false);
when(executor.awaitTermination(10, TimeUnit.MILLISECONDS)).thenReturn(false, true);
ExecutorUtil.gracefulShutdown(executor, 20);
Thread.sleep(2000);
verify(executor).shutdown();
verify(executor, atLeast(2)).shutdownNow();
await().untilAsserted(() -> verify(executor, times(2)).awaitTermination(10, TimeUnit.MILLISECONDS));
verify(executor, times(1)).shutdown();
verify(executor, times(3)).shutdownNow();
}
@Test
@ -82,4 +85,4 @@ class ExecutorUtilTest {
url = ExecutorUtil.setThreadName(url, "default-name");
assertThat(url.getParameter(THREAD_NAME_KEY), equalTo("custom-thread-localhost:1234"));
}
}
}

4
dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/ServiceConfigTest.java
View File

@ -232,7 +232,7 @@ class ServiceConfigTest {
try {
service.export();
service.unexport();
Thread.sleep(1000);
// Thread.sleep(1000);
Mockito.verify(exporter, Mockito.atLeastOnce()).unexport();
} finally {
System.clearProperty(SHUTDOWN_TIMEOUT_KEY);
@ -529,4 +529,4 @@ class ServiceConfigTest {
service.export();
});
}
}
}

36
dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/beans/factory/annotation/MethodConfigCallbackTest.java
View File

@ -39,13 +39,15 @@ import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import static org.awaitility.Awaitility.await;
@ExtendWith(SpringExtension.class)
@ContextConfiguration(
classes = {
ProviderConfiguration.class,
MethodConfigCallbackTest.class,
MethodConfigCallbackTest.MethodCallbackConfiguration.class
})
classes = {
ProviderConfiguration.class,
MethodConfigCallbackTest.class,
MethodConfigCallbackTest.MethodCallbackConfiguration.class
})
@TestPropertySource(properties = {
"dubbo.protocol.port=-1",
"dubbo.registry.address=${zookeeper.connection.address}"
@ -70,14 +72,14 @@ class MethodConfigCallbackTest {
@DubboReference(check = false, async = true,
injvm = false, // Currently, local call is not supported method callback cause by Injvm protocol is not supported ClusterFilter
methods = {@Method(name = "sayHello",
oninvoke = "methodCallback.oninvoke1",
onreturn = "methodCallback.onreturn1",
onthrow = "methodCallback.onthrow1")})
oninvoke = "methodCallback.oninvoke1",
onreturn = "methodCallback.onreturn1",
onthrow = "methodCallback.onthrow1")})
private HelloService helloServiceMethodCallBack;
@DubboReference(check = false, async = true,
injvm = false, // Currently, local call is not supported method callback cause by Injvm protocol is not supported ClusterFilter
methods = {@Method(name = "sayHello",
injvm = false, // Currently, local call is not supported method callback cause by Injvm protocol is not supported ClusterFilter
methods = {@Method(name = "sayHello",
oninvoke = "methodCallback.oninvoke2",
onreturn = "methodCallback.onreturn2",
onthrow = "methodCallback.onthrow2")})
@ -95,21 +97,13 @@ class MethodConfigCallbackTest {
}
}).start();
}
int i = 0;
while (MethodCallbackImpl.cnt.get() < ( 2 * threadCnt * callCnt)){
// wait for async callback finished
try {
i++;
Thread.sleep(100);
} catch (InterruptedException e) {
}
}
await().until(() -> MethodCallbackImpl.cnt.get() >= (2 * threadCnt * callCnt));
MethodCallback notify = (MethodCallback) context.getBean("methodCallback");
StringBuilder invoke1Builder = new StringBuilder();
StringBuilder invoke2Builder = new StringBuilder();
StringBuilder return1Builder = new StringBuilder();
StringBuilder return2Builder = new StringBuilder();
for (i = 0; i < threadCnt * callCnt; i++) {
for (int i = 0; i < threadCnt * callCnt; i++) {
invoke1Builder.append("dubbo invoke success!");
invoke2Builder.append("dubbo invoke success(2)!");
return1Builder.append("dubbo return success!");
@ -130,4 +124,4 @@ class MethodConfigCallbackTest {
}
}
}
}

2
dubbo-metadata/dubbo-metadata-report-zookeeper/src/main/java/org/apache/dubbo/metadata/store/zookeeper/ZookeeperMetadataReport.java
View File

@ -198,7 +198,7 @@ public class ZookeeperMetadataReport extends AbstractMetadataReport {
throw new IllegalArgumentException("zookeeper publishConfigCas requires stat type ticket");
}
String pathKey = buildPathKey(group, key);
zkClient.createOrUpdate(pathKey, content, false, ticket == null ? 0 : ((Stat) ticket).getVersion());
zkClient.createOrUpdate(pathKey, content, false, ticket == null ? null : ((Stat) ticket).getVersion());
return true;
} catch (Exception e) {
logger.warn(REGISTRY_ZOOKEEPER_EXCEPTION, "", "", "zookeeper publishConfigCas failed.", e);

6
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/zookeeper/AbstractZookeeperClient.java
View File

@ -178,7 +178,7 @@ public abstract class AbstractZookeeperClient<TargetDataListener, TargetChildLis
}
@Override
public void createOrUpdate(String path, String content, boolean ephemeral, int version) {
public void createOrUpdate(String path, String content, boolean ephemeral, Integer version) {
int i = path.lastIndexOf('/');
if (i > 0) {
create(path.substring(0, i), false, true);
@ -224,9 +224,9 @@ public abstract class AbstractZookeeperClient<TargetDataListener, TargetChildLis
protected abstract void createOrUpdateEphemeral(String path, String data);
protected abstract void createOrUpdatePersistent(String path, String data, int version);
protected abstract void createOrUpdatePersistent(String path, String data, Integer version);
protected abstract void createOrUpdateEphemeral(String path, String data, int version);
protected abstract void createOrUpdateEphemeral(String path, String data, Integer version);
@Override
public abstract boolean checkExists(String path);

2
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/zookeeper/ZookeeperClient.java
View File

@ -110,7 +110,7 @@ public interface ZookeeperClient {
* @param ephemeral specify create mode of ZNode creation. true - EPHEMERAL, false - PERSISTENT.
* @param ticket origin content version, if current version is not the specified version, throw exception
*/
void createOrUpdate(String path, String content, boolean ephemeral, int ticket);
void createOrUpdate(String path, String content, boolean ephemeral, Integer ticket);
/**
* Obtain the content of a ZNode.

8
dubbo-remoting/dubbo-remoting-zookeeper-curator5/src/main/java/org/apache/dubbo/remoting/zookeeper/curator5/Curator5ZookeeperClient.java
View File

@ -240,9 +240,9 @@ public class Curator5ZookeeperClient extends AbstractZookeeperClient<Curator5Zoo
}
@Override
protected void createOrUpdatePersistent(String path, String data, int version) {
protected void createOrUpdatePersistent(String path, String data, Integer version) {
try {
if (checkExists(path)) {
if (checkExists(path) && version != null) {
update(path, data, version);
} else {
createPersistent(path, data, false);
@ -253,9 +253,9 @@ public class Curator5ZookeeperClient extends AbstractZookeeperClient<Curator5Zoo
}
@Override
protected void createOrUpdateEphemeral(String path, String data, int version) {
protected void createOrUpdateEphemeral(String path, String data, Integer version) {
try {
if (checkExists(path)) {
if (checkExists(path) && version != null) {
update(path, data, version);
} else {
createEphemeral(path, data, false);

34
dubbo-remoting/dubbo-remoting-zookeeper-curator5/src/test/java/org/apache/dubbo/remoting/zookeeper/curator5/Curator5ZookeeperClientTest.java
View File

@ -223,7 +223,8 @@ class Curator5ZookeeperClientTest {
}
@Test
void testPersistentCas() throws Exception {
void testPersistentCas1() throws Exception {
// test create failed when others create success
String path = "/dubbo/mapping/org.apache.dubbo.demo.DemoService";
AtomicReference<Runnable> runnable = new AtomicReference<>();
Curator5ZookeeperClient curatorClient = new Curator5ZookeeperClient(URL.valueOf(zookeeperConnectionAddress1 + "/org.apache.dubbo.registry.RegistryService")) {
@ -278,6 +279,20 @@ class Curator5ZookeeperClientTest {
curatorClient.close();
}
@Test
void testPersistentCas2() throws Exception {
// test update failed when others create success
String path = "/dubbo/mapping/org.apache.dubbo.demo.DemoService";
Curator5ZookeeperClient curatorClient = new Curator5ZookeeperClient(URL.valueOf(zookeeperConnectionAddress1 + "/org.apache.dubbo.registry.RegistryService"));
curatorClient.delete(path);
curatorClient.createOrUpdate(path, "version x", false);
Assertions.assertThrows(IllegalStateException.class, () -> curatorClient.createOrUpdate(path, "version 1", false, null));
Assertions.assertEquals("version x", curatorClient.getContent(path));
curatorClient.close();
}
@Test
void testPersistentNonVersion() {
String path = "/dubbo/metadata/org.apache.dubbo.demo.DemoService";
@ -333,7 +348,8 @@ class Curator5ZookeeperClientTest {
}
@Test
void testEphemeralCas() throws Exception {
void testEphemeralCas1() throws Exception {
// test create failed when others create success
String path = "/dubbo/mapping/org.apache.dubbo.demo.DemoService";
AtomicReference<Runnable> runnable = new AtomicReference<>();
Curator5ZookeeperClient curatorClient = new Curator5ZookeeperClient(URL.valueOf(zookeeperConnectionAddress1 + "/org.apache.dubbo.registry.RegistryService")) {
@ -388,6 +404,20 @@ class Curator5ZookeeperClientTest {
curatorClient.close();
}
@Test
void testEphemeralCas2() throws Exception {
// test update failed when others create success
String path = "/dubbo/mapping/org.apache.dubbo.demo.DemoService";
Curator5ZookeeperClient curatorClient = new Curator5ZookeeperClient(URL.valueOf(zookeeperConnectionAddress1 + "/org.apache.dubbo.registry.RegistryService"));
curatorClient.delete(path);
curatorClient.createOrUpdate(path, "version x", true);
Assertions.assertThrows(IllegalStateException.class, () -> curatorClient.createOrUpdate(path, "version 1", true, null));
Assertions.assertEquals("version x", curatorClient.getContent(path));
curatorClient.close();
}
@Test
void testEphemeralNonVersion() {
String path = "/dubbo/metadata/org.apache.dubbo.demo.DemoService";

8
dubbo-remoting/dubbo-remoting-zookeeper/src/main/java/org/apache/dubbo/remoting/zookeeper/curator/CuratorZookeeperClient.java
View File

@ -244,9 +244,9 @@ public class CuratorZookeeperClient extends AbstractZookeeperClient<CuratorZooke
}
@Override
protected void createOrUpdatePersistent(String path, String data, int version) {
protected void createOrUpdatePersistent(String path, String data, Integer version) {
try {
if (checkExists(path)) {
if (checkExists(path) && version != null) {
update(path, data, version);
} else {
createPersistent(path, data, false);
@ -257,9 +257,9 @@ public class CuratorZookeeperClient extends AbstractZookeeperClient<CuratorZooke
}
@Override
protected void createOrUpdateEphemeral(String path, String data, int version) {
protected void createOrUpdateEphemeral(String path, String data, Integer version) {
try {
if (checkExists(path)) {
if (checkExists(path) && version != null) {
update(path, data, version);
} else {
createEphemeral(path, data, false);

34
dubbo-remoting/dubbo-remoting-zookeeper/src/test/java/org/apache/dubbo/remoting/zookeeper/curator/CuratorZookeeperClientTest.java
View File

@ -248,7 +248,8 @@ class CuratorZookeeperClientTest {
@Test
void testPersistentCas() throws Exception {
void testPersistentCas1() throws Exception {
// test create failed when others create success
String path = "/dubbo/mapping/org.apache.dubbo.demo.DemoService";
AtomicReference<Runnable> runnable = new AtomicReference<>();
CuratorZookeeperClient curatorClient = new CuratorZookeeperClient(URL.valueOf(zookeeperConnectionAddress1 + "/org.apache.dubbo.registry.RegistryService")) {
@ -303,6 +304,20 @@ class CuratorZookeeperClientTest {
curatorClient.close();
}
@Test
void testPersistentCas2() throws Exception {
// test update failed when others create success
String path = "/dubbo/mapping/org.apache.dubbo.demo.DemoService";
CuratorZookeeperClient curatorClient = new CuratorZookeeperClient(URL.valueOf(zookeeperConnectionAddress1 + "/org.apache.dubbo.registry.RegistryService"));
curatorClient.delete(path);
curatorClient.createOrUpdate(path, "version x", false);
Assertions.assertThrows(IllegalStateException.class, () -> curatorClient.createOrUpdate(path, "version 1", false, null));
Assertions.assertEquals("version x", curatorClient.getContent(path));
curatorClient.close();
}
@Test
void testPersistentNonVersion() {
String path = "/dubbo/metadata/org.apache.dubbo.demo.DemoService";
@ -358,7 +373,8 @@ class CuratorZookeeperClientTest {
}
@Test
void testEphemeralCas() throws Exception {
void testEphemeralCas1() throws Exception {
// test create failed when others create success
String path = "/dubbo/mapping/org.apache.dubbo.demo.DemoService";
AtomicReference<Runnable> runnable = new AtomicReference<>();
CuratorZookeeperClient curatorClient = new CuratorZookeeperClient(URL.valueOf(zookeeperConnectionAddress1 + "/org.apache.dubbo.registry.RegistryService")) {
@ -413,6 +429,20 @@ class CuratorZookeeperClientTest {
curatorClient.close();
}
@Test
void testEphemeralCas2() throws Exception {
// test update failed when others create success
String path = "/dubbo/mapping/org.apache.dubbo.demo.DemoService";
CuratorZookeeperClient curatorClient = new CuratorZookeeperClient(URL.valueOf(zookeeperConnectionAddress1 + "/org.apache.dubbo.registry.RegistryService"));
curatorClient.delete(path);
curatorClient.createOrUpdate(path, "version x", true);
Assertions.assertThrows(IllegalStateException.class, () -> curatorClient.createOrUpdate(path, "version 1", true, null));
Assertions.assertEquals("version x", curatorClient.getContent(path));
curatorClient.close();
}
@Test
void testEphemeralNonVersion() {
String path = "/dubbo/metadata/org.apache.dubbo.demo.DemoService";

21
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/support/RpcUtils.java
View File

@ -261,26 +261,21 @@ public class RpcUtils {
return timeout;
}
public static long getTimeout(URL url, String methodName, RpcContext context, long defaultTimeout) {
public static long getTimeout(URL url, String methodName, RpcContext context, Invocation invocation, long defaultTimeout) {
long timeout = defaultTimeout;
Object genericTimeout = context.getObjectAttachment(TIMEOUT_KEY);
if (genericTimeout != null) {
timeout = convertToNumber(genericTimeout, defaultTimeout);
Object timeoutFromContext = context.getObjectAttachment(TIMEOUT_KEY);
Object timeoutFromInvocation = invocation.getObjectAttachment(TIMEOUT_KEY);
if (timeoutFromContext != null) {
timeout = convertToNumber(timeoutFromContext, defaultTimeout);
} else if (timeoutFromInvocation != null) {
timeout = convertToNumber(timeoutFromInvocation, defaultTimeout);
} else if (url != null) {
timeout = url.getMethodPositiveParameter(methodName, TIMEOUT_KEY, defaultTimeout);
}
return timeout;
}
public static long getTimeoutFromInvocation(Invocation invocation, long defaultTimeout) {
long timeout = defaultTimeout;
Object genericTimeout = invocation.getObjectAttachment(TIMEOUT_KEY);
if (genericTimeout != null) {
timeout = convertToNumber(genericTimeout, defaultTimeout);
}
return timeout;
}
private static long convertToNumber(Object obj, long defaultTimeout) {
long timeout = defaultTimeout;
try {

2
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodec.java
View File

@ -159,7 +159,7 @@ public class DubboCodec extends ExchangeCodec {
req.setData(data);
} catch (Throwable t) {
if (log.isWarnEnabled()) {
log.warn("Decode request failed: " + t.getMessage(), t);
log.warn(PROTOCOL_FAILED_DECODE, "", "", "Decode request failed: " + t.getMessage(), t);
}
// bad request
req.setBroken(true);

2
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvoker.java
View File

@ -178,7 +178,7 @@ public class DubboInvoker<T> extends AbstractInvoker<T> {
Object countdown = RpcContext.getClientAttachment().getObjectAttachment(TIME_COUNTDOWN_KEY);
int timeout;
if (countdown == null) {
timeout = (int) RpcUtils.getTimeout(getUrl(), methodName, RpcContext.getClientAttachment(), DEFAULT_TIMEOUT);
timeout = (int) RpcUtils.getTimeout(getUrl(), methodName, RpcContext.getClientAttachment(), invocation, DEFAULT_TIMEOUT);
if (getUrl().getParameter(ENABLE_TIMEOUT_COUNTDOWN_KEY, false)) {
invocation.setObjectAttachment(TIMEOUT_ATTACHMENT_KEY, timeout); // pass timeout to remote server
}

2
dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/InjvmInvoker.java
View File

@ -283,7 +283,7 @@ public class InjvmInvoker<T> extends AbstractInvoker<T> {
Object countdown = RpcContext.getClientAttachment().getObjectAttachment(TIME_COUNTDOWN_KEY);
int timeout;
if (countdown == null) {
timeout = (int) RpcUtils.getTimeout(getUrl(), methodName, RpcContext.getClientAttachment(), DEFAULT_TIMEOUT);
timeout = (int) RpcUtils.getTimeout(getUrl(), methodName, RpcContext.getClientAttachment(), invocation, DEFAULT_TIMEOUT);
if (getUrl().getParameter(ENABLE_TIMEOUT_COUNTDOWN_KEY, false)) {
invocation.setObjectAttachment(TIMEOUT_ATTACHMENT_KEY, timeout); // pass timeout to remote server
}

7
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java
View File

@ -106,7 +106,7 @@ public class TripleInvoker<T> extends AbstractInvoker<T> {
if (!connectionClient.isConnected()) {
CompletableFuture<AppResponse> future = new CompletableFuture<>();
RpcException exception = TriRpcStatus.UNAVAILABLE.withDescription(
String.format("upstream %s is unavailable", getUrl().getAddress()))
String.format("upstream %s is unavailable", getUrl().getAddress()))
.asException();
future.completeExceptionally(exception);
return new AsyncRpcResult(future, invocation);
@ -301,14 +301,11 @@ public class TripleInvoker<T> extends AbstractInvoker<T> {
}
private int calculateTimeout(Invocation invocation, String methodName) {
if (invocation.getObjectAttachment(TIMEOUT_KEY) != null) {
return (int) RpcUtils.getTimeoutFromInvocation(invocation, 3000);
}
Object countdown = RpcContext.getClientAttachment().getObjectAttachment(TIME_COUNTDOWN_KEY);
int timeout;
if (countdown == null) {
timeout = (int) RpcUtils.getTimeout(getUrl(), methodName,
RpcContext.getClientAttachment(), 3000);
RpcContext.getClientAttachment(), invocation, 3000);
if (getUrl().getParameter(ENABLE_TIMEOUT_COUNTDOWN_KEY, false)) {
invocation.setObjectAttachment(TIMEOUT_ATTACHMENT_KEY,
timeout); // pass timeout to remote server