Skip to content

Commit 501f2a4

Browse files
coderbruiscoderbruis
committed
fix
1 parent 3dbae34 commit 501f2a4

3 files changed

Lines changed: 192 additions & 22 deletions

File tree

‎JdkLearn/src/main/java/com/learnjava/concurent/CompletableFutureDemo.java‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,8 @@ public class CompletableFutureDemo {
66

77
public static void main(String[] args) throws Exception {
88
CompletableFutureDemo completableFutureDemo = new CompletableFutureDemo();
9-
completableFutureDemo.test_completed_future();
9+
// completableFutureDemo.test_completed_future();
10+
completableFutureDemo.test_completed_thenApply();
1011
}
1112

1213
public void test_completed_future() throws Exception {
@@ -15,4 +16,14 @@ public void test_completed_future() throws Exception {
1516
System.out.println(alreadyCompleted.get());
1617
}
1718

19+
public void test_completed_thenApply() {
20+
CompletableFuture<Integer> source = new CompletableFuture<>();
21+
CompletableFuture<Integer> next =
22+
source.thenApply(x -> {
23+
System.out.println(Thread.currentThread().getName());
24+
return x + 1;
25+
});
26+
source.complete(10);
27+
System.out.println(next.join()); // 11
28+
}
1829
}

‎JdkLearn/src/test/java/com/learnjava/concurrent/CompletableFuturePrincipleTest.java‎

Lines changed: 159 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,159 @@
1+
package com.learnjava.concurrent;
2+
3+
import org.junit.jupiter.api.Test;
4+
import org.junit.jupiter.api.Timeout;
5+
6+
import java.util.concurrent.CompletableFuture;
7+
import java.util.concurrent.CompletionException;
8+
import java.util.concurrent.CountDownLatch;
9+
import java.util.concurrent.ExecutorService;
10+
import java.util.concurrent.Executors;
11+
import java.util.concurrent.TimeUnit;
12+
import java.util.concurrent.atomic.AtomicBoolean;
13+
import java.util.concurrent.atomic.AtomicReference;
14+
15+
import static org.junit.jupiter.api.Assertions.assertEquals;
16+
import static org.junit.jupiter.api.Assertions.assertFalse;
17+
import static org.junit.jupiter.api.Assertions.assertSame;
18+
import static org.junit.jupiter.api.Assertions.assertThrows;
19+
import static org.junit.jupiter.api.Assertions.assertTrue;;
20+
21+
/**
22+
* @Author : haiyang.luo
23+
* @Date : 2026/6/28 21:44
24+
* @Description :
25+
*/
26+
@Timeout(30)
27+
public class CompletableFuturePrincipleTest {
28+
29+
@Test
30+
public void testCompleteTriggersDependentStageRegisteredBeforeCompletion() {
31+
CompletableFuture<Integer> source = new CompletableFuture<>();
32+
CompletableFuture<Integer> dependent = source.thenApply(value -> value + 1);
33+
34+
assertFalse(dependent.isDone());
35+
36+
assertTrue(source.complete(10));
37+
38+
assertEquals(11, dependent.join());
39+
assertTrue(dependent.isDone());
40+
}
41+
42+
@Test
43+
public void testDependentStageRegisteredAfterCompletionRunsOnCallerThread() {
44+
CompletableFuture<Integer> source = CompletableFuture.completedFuture(10);
45+
String callerThreadName = Thread.currentThread().getName();
46+
AtomicReference<String> callbackThreadName = new AtomicReference<>();
47+
48+
CompletableFuture<Integer> dependent = source.thenApply(value -> {
49+
callbackThreadName.set(Thread.currentThread().getName());
50+
return value + 1;
51+
});
52+
53+
assertEquals(11, dependent.join());
54+
assertEquals(callerThreadName, callbackThreadName.get());
55+
}
56+
57+
@Test
58+
public void testAsyncDependentStageRunsOnExecutor() {
59+
ExecutorService executor = Executors.newSingleThreadExecutor(runnable -> {
60+
Thread thread = new Thread(runnable);
61+
thread.setName("completable-future-principle-test");
62+
return thread;
63+
});
64+
try {
65+
CompletableFuture<Integer> source = new CompletableFuture<>();
66+
AtomicReference<String> callbackThreadName = new AtomicReference<>();
67+
68+
CompletableFuture<Integer> dependent = source.thenApplyAsync(value -> {
69+
callbackThreadName.set(Thread.currentThread().getName());
70+
return value + 1;
71+
}, executor);
72+
73+
assertTrue(source.complete(10));
74+
75+
assertEquals(11, dependent.join());
76+
assertEquals("completable-future-principle-test", callbackThreadName.get());
77+
} finally {
78+
executor.shutdownNow();
79+
}
80+
}
81+
82+
@Test
83+
public void testThenComposeCompletesAfterReturnedFutureCompletes() {
84+
CompletableFuture<Integer> source = new CompletableFuture<>();
85+
CompletableFuture<Integer> inner = new CompletableFuture<>();
86+
87+
CompletableFuture<Integer> composed = source.thenCompose(value -> inner.thenApply(innerValue -> value + innerValue));
88+
89+
assertTrue(source.complete(10));
90+
assertFalse(composed.isDone());
91+
92+
assertTrue(inner.complete(5));
93+
94+
assertEquals(15, composed.join());
95+
}
96+
97+
@Test
98+
public void testExceptionalCompletionSkipsThenApplyAndCanRecover() {
99+
CompletableFuture<Integer> source = new CompletableFuture<>();
100+
AtomicBoolean thenApplyCalled = new AtomicBoolean(false);
101+
102+
CompletableFuture<Integer> recovered = source
103+
.thenApply(value -> {
104+
thenApplyCalled.set(true);
105+
return value + 1;
106+
})
107+
.exceptionally(exception -> 0);
108+
109+
RuntimeException failure = new RuntimeException("boom");
110+
assertTrue(source.completeExceptionally(failure));
111+
112+
assertEquals(0, recovered.join());
113+
assertFalse(thenApplyCalled.get());
114+
}
115+
116+
@Test
117+
public void testOnlyFirstCompletionWins() {
118+
CompletableFuture<String> future = new CompletableFuture<>();
119+
120+
assertTrue(future.complete("first"));
121+
assertFalse(future.complete("second"));
122+
assertFalse(future.completeExceptionally(new RuntimeException("boom")));
123+
124+
assertEquals("first", future.join());
125+
}
126+
127+
@Test
128+
public void testJoinWaitsForCompletion() throws Exception {
129+
CompletableFuture<String> future = new CompletableFuture<>();
130+
CountDownLatch joinStarted = new CountDownLatch(1);
131+
AtomicReference<String> joinedValue = new AtomicReference<>();
132+
133+
Thread waitingThread = new Thread(() -> {
134+
joinStarted.countDown();
135+
joinedValue.set(future.join());
136+
});
137+
waitingThread.start();
138+
139+
assertTrue(joinStarted.await(5, TimeUnit.SECONDS));
140+
assertFalse(future.isDone());
141+
142+
assertTrue(future.complete("done"));
143+
144+
waitingThread.join(TimeUnit.SECONDS.toMillis(5));
145+
assertFalse(waitingThread.isAlive());
146+
assertEquals("done", joinedValue.get());
147+
}
148+
149+
@Test
150+
public void testJoinWrapsFailureInCompletionException() {
151+
CompletableFuture<String> future = new CompletableFuture<>();
152+
RuntimeException failure = new RuntimeException("boom");
153+
154+
assertTrue(future.completeExceptionally(failure));
155+
156+
CompletionException exception = assertThrows(CompletionException.class, future::join);
157+
assertSame(failure, exception.getCause());
158+
}
159+
}

‎note/kafka/kafka rebalance核心逻辑分析.md‎

Lines changed: 21 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -3,15 +3,15 @@
33
# Kafka Rebalance 核心流程
44
`KafkaConsumer.poll()` 是消费者触发 rebalance 的主要入口。
55

6-
位置在 [KafkaConsumer.java (line 916)](/Users/apple/Documents/git_code_pr/kafka/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:916)：
6+
位置在KafkaConsumer.java：
77

88
```plain
99
public ConsumerRecords<K, V> poll(final Duration timeout) {
1010
return delegate.poll(timeout);
1111
}
1212
```
1313

14-
在 classic consumer 实现里，会进入 [ClassicKafkaConsumer.poll() (line 641)](/Users/apple/Documents/git_code_pr/kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java:641)：
14+
在 classic consumer 实现里，会进入 ClassicKafkaConsumer.poll()：
1515

1616
```plain
1717
KafkaConsumer.poll
@@ -33,22 +33,22 @@ KafkaConsumer.poll
3333
+ **拉取并返回消息**
3434
从 fetch buffer 取数据，或者发送新的 FetchRequest，并把拿到的 records 返回给用户。
3535

36-
对应 [ClassicKafkaConsumer.java (line 665)](/Users/apple/Documents/git_code_pr/kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java:665)：
36+
对应 ClassicKafkaConsumer.java：
3737

3838
```plain
3939
updateAssignmentMetadataIfNeeded(timer, false)
4040
pollForFetches(timer)
4141
sendFetches()
4242
```
4343

44-
再精简一句：**rebalance 不是后台自动完整完成的，classic consumer 的 group rebalance 主要是在用户线程调用 **`poll()`** 时被推进的。**
44+
**rebalance 不是后台自动完整完成的，classic consumer 的 group rebalance 主要是在用户线程调用 **`poll()`** 时被推进的。**
4545

4646
---
4747

4848
## ConsumerCoordinator.poll()
4949
`ConsumerCoordinator.poll()` 是 consumer 端消费组协调的入口。
5050

51-
位置在 [ConsumerCoordinator.java (line 530)](/Users/apple/Documents/git_code_pr/kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:530)：
51+
位置在 ConsumerCoordinator.java：
5252

5353
```plain
5454
public boolean poll(Timer timer, boolean waitForJoinGroup)
@@ -105,7 +105,7 @@ Kafka 不希望在内部网络处理路径里直接跑用户 callback，因为
105105

106106

107107
### rejoinNeededOrPending（）
108-
这个方法命名就很有意思“是否需要重新加入消费组，或者重新加入消费组的流程仍在进行中”。首先来看请求第一次进来。
108+
这个方法命名就很有意思“是否需要重新加入消费组，或者重新加入消费组的流程仍在进行中”。首先来看请求第一次进来。
109109

110110
核心链路：
111111

@@ -127,7 +127,7 @@ if (!subscriptions.hasAutoAssignedPartitions())
127127

128128

129129

130-
上次 assignment 使用的 metadata 和当前 metadata 不匹配，或者说集群 topic/partition 等元数据变化，需要重新分配。
130+
上次 assignment 使用的 metadata 和当前 metadata 不匹配，或者说集群 topic/partition 等元数据变化，需要重新分配。
131131

132132
```java
133133
if (assignmentSnapshot != null && !assignmentSnapshot.matches(metadataSnapshot)) {
@@ -138,7 +138,7 @@ if (assignmentSnapshot != null && !assignmentSnapshot.matches(metadataSnapshot))
138138

139139

140140

141-
上次 JoinGroup 提交的 subscription 和当前 subscription 不一致，或者说用户订阅或正则匹配结果变化，需要重新入组。
141+
上次 JoinGroup 提交的 subscription 和当前 subscription 不一致，或者说用户订阅或正则匹配结果变化，需要重新入组。
142142

143143
```java
144144
if (joinedSubscription != null && !joinedSubscription.equals(subscriptions.subscription())) {
@@ -149,7 +149,7 @@ if (joinedSubscription != null && !joinedSubscription.equals(subscriptions.subsc
149149

150150

151151

152-
这里两处判断为啥都是上次的JoinGroup？因为这是在构造这一次JoinGroup之前，所以得那上次的JoinGroup的元数据和订阅信息和当前准备构造JoinGroup的数据进行对比。
152+
这里两处判断为啥都是上次的JoinGroup？因为这是在构造当前这次JoinGroup之前的判断逻辑，所以得拿上次的JoinGroup的元数据和订阅信息和当前准备构造JoinGroup的数据进行对比。
153153

154154

155155

@@ -179,7 +179,7 @@ public synchronized void requestRejoin(final String shortReason,
179179
## Rebalance触发条件？
180180
是否需要加入或重新加入消费组，主要由 `rejoinNeededOrPending()` 判断。
181181

182-
位置在 [ConsumerCoordinator.java (line 953)](/Users/apple/Documents/git_code_pr/kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:953)：
182+
位置在 ConsumerCoordinator.java：
183183

184184
```plain
185185
public boolean rejoinNeededOrPending()
@@ -203,7 +203,7 @@ topic 新增分区、删除分区、正则订阅匹配结果变化等。
203203
## ensureActiveGroup()
204204
`ensureActiveGroup()` 是“确保消费组已经可用”的核心方法。
205205

206-
位置在 [AbstractCoordinator.java (line 422)](/Users/apple/Documents/git_code_pr/kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:422)：
206+
位置在 AbstractCoordinator.java：
207207

208208
```plain
209209
boolean ensureActiveGroup(final Timer timer) {
@@ -240,7 +240,7 @@ ensureActiveGroup
240240
## joinGroupIfNeeded()
241241
`joinGroupIfNeeded()` 是 consumer 端 rebalance 状态机的核心。
242242

243-
位置在 [AbstractCoordinator.java (line 476)](/Users/apple/Documents/git_code_pr/kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:476)：
243+
位置在 AbstractCoordinator.java：
244244

245245
```plain
246246
boolean joinGroupIfNeeded(final Timer timer)
@@ -280,7 +280,7 @@ SyncGroup 成功后，调用 `onJoinComplete(...)`，更新本地分配并触发
280280
## JoinGroup 请求携带什么
281281
JoinGroup 请求里最关键的是 `metadata()` 生成的 subscription metadata。
282282

283-
位置在 [ConsumerCoordinator.java (line 303)](/Users/apple/Documents/git_code_pr/kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:303)：
283+
位置在 ConsumerCoordinator.java：
284284

285285
```plain
286286
protected JoinGroupRequestProtocolCollection metadata()
@@ -313,7 +313,7 @@ Subscription subscription = new Subscription(
313313
## Broker 侧 JoinGroup / SyncGroup
314314
broker 网络入口仍然是 `KafkaApis.handle()`。
315315

316-
位置在 [KafkaApis.scala (line 180)](/Users/apple/Documents/git_code_pr/kafka/core/src/main/scala/kafka/server/KafkaApis.scala:180)：
316+
位置在 KafkaApis.scala：
317317

318318
```plain
319319
case ApiKeys.JOIN_GROUP => handleJoinGroupRequest(request, requestLocal).exceptionally(handleError)
@@ -322,7 +322,7 @@ case ApiKeys.SYNC_GROUP => handleSyncGroupRequest(request, requestLocal).excepti
322322

323323
`handleJoinGroupRequest()` 会做 group 权限校验，然后交给 `groupCoordinator.joinGroup()`。
324324

325-
位置在 [KafkaApis.scala (line 1381)](/Users/apple/Documents/git_code_pr/kafka/core/src/main/scala/kafka/server/KafkaApis.scala:1381)：
325+
位置在 KafkaApis.scala：
326326

327327
```plain
328328
groupCoordinator.joinGroup(
@@ -334,7 +334,7 @@ groupCoordinator.joinGroup(
334334

335335
`GroupCoordinatorService.joinGroup()` 最终调度 classic group 的 join 操作。
336336

337-
位置在 [GroupCoordinatorService.java (line 1053)](/Users/apple/Documents/git_code_pr/kafka/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:1053)：
337+
位置在 GroupCoordinatorService.java ：
338338

339339
```plain
340340
runtime.scheduleWriteOperation(
@@ -346,7 +346,7 @@ runtime.scheduleWriteOperation(
346346

347347
`syncGroup()` 类似，会调度 classic group sync。
348348

349-
位置在 [GroupCoordinatorService.java (line 1125)](/Users/apple/Documents/git_code_pr/kafka/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:1125)：
349+
位置在 GroupCoordinatorService.java：
350350

351351
```plain
352352
runtime.scheduleWriteOperation(
@@ -377,7 +377,7 @@ JoinGroupResponseHandler.handle
377377
-> ConsumerCoordinator.onLeaderElected(...)
378378
```
379379

380-
位置在 [AbstractCoordinator.java (line 724)](/Users/apple/Documents/git_code_pr/kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:724)：
380+
位置在 AbstractCoordinator.java：
381381

382382
```plain
383383
if (joinResponse.isLeader()) {
@@ -387,7 +387,7 @@ if (joinResponse.isLeader()) {
387387
}
388388
```
389389

390-
leader 的核心逻辑在 [ConsumerCoordinator.onLeaderElected() (line 686)](/Users/apple/Documents/git_code_pr/kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:686)：
390+
leader 的核心逻辑在 ConsumerCoordinator.onLeaderElected()：
391391

392392
```plain
393393
protected Map<String, ByteBuffer> onLeaderElected(...)
@@ -408,14 +408,14 @@ protected Map<String, ByteBuffer> onLeaderElected(...)
408408
+ **序列化 assignment**
409409
把每个 member 的 assignment 序列化，准备放进 leader SyncGroup。
410410

411-
核心代码在 [ConsumerCoordinator.java (line 724)](/Users/apple/Documents/git_code_pr/kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:724)：
411+
核心代码在 ConsumerCoordinator.java：
412412

413413
```plain
414414
Map<String, Assignment> assignments =
415415
assignor.assign(metadata.fetch(), new GroupSubscription(subscriptions)).groupAssignment();
416416
```
417417

418-
leader 计算完 assignment 后，会在 [AbstractCoordinator.java (line 836)](/Users/apple/Documents/git_code_pr/kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:836) 构造 leader SyncGroup：
418+
leader 计算完 assignment 后，会在 AbstractCoordinator.java 构造 leader SyncGroup：
419419

420420
```plain
421421
SyncGroupRequest.Builder requestBuilder =

0 commit comments

Comments
 (0)