This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 3ae637febe perf(admin): coalesce burst HTTP cache rebuilds by
namespace and group (#7843)
3ae637febe is described below
commit 3ae637febe8245e447d6e27471d4d1af70ea3fcc
Author: redyyang <[email protected]>
AuthorDate: Tue Oct 20 12:37:54 2026 +0800
perf(admin): coalesce burst HTTP cache rebuilds by namespace and group
(#7843)
* perf(admin): coalesce HTTP cache rebuilds by namespace and group
* perf(admin): parallelize, back off and degrade gracefully in HTTP cache
rebuild coalescing
Address review feedback on #7843:
- Add shenyu.sync.http.cache-rebuild-threads (default 1, 1-64). Each key
keeps a single scheduled or running drain, so distinct namespace/group
keys
rebuild in parallel while one key never rebuilds concurrently. Document
the
single-worker serialization trade-off.
- Back off failed rebuilds exponentially (1s doubling, capped at 30s, never
below the window) and reset after a success, instead of retrying every
second forever. Retries still never give up, so the newest committed
configuration is eventually published.
- Return false from request() after close so a commit racing shutdown
rebuilds synchronously instead of failing its after-commit hook; skip the
long-poll notification once the notification scheduler has stopped.
- Explain why notifyChanged() is a no-op when coalescing is enabled.
Refs #7839
---
docs/benchmarks/http-cache-rebuild-256.csv | 73 ++++
docs/benchmarks/http-cache-rebuild-4096.csv | 73 ++++
docs/http-cache-rebuild-coalescing.md | 222 +++++++++++
.../config/properties/HttpSyncProperties.java | 82 ++++
.../listener/AbstractDataChangedListener.java | 26 +-
.../listener/http/HttpCacheRebuildCoordinator.java | 173 +++++++++
.../http/HttpLongPollingDataChangedListener.java | 83 +++-
shenyu-admin/src/main/resources/application.yml | 3 +
.../config/properties/HttpSyncPropertiesTest.java | 20 +
.../listener/http/HttpCacheRebuildBenchmark.java | 425 +++++++++++++++++++++
.../http/HttpCacheRebuildCoordinatorTest.java | 296 ++++++++++++++
.../admin/listener/http/HttpCacheRebuildTest.java | 357 +++++++++++++++++
12 files changed, 1818 insertions(+), 15 deletions(-)
diff --git a/docs/benchmarks/http-cache-rebuild-256.csv
b/docs/benchmarks/http-cache-rebuild-256.csv
new file mode 100644
index 0000000000..abeae94b56
--- /dev/null
+++ b/docs/benchmarks/http-cache-rebuild-256.csv
@@ -0,0 +1,73 @@
+namespaces,groups,pace_ms,window_ms,sample,rows_per_key,handle_chars,events,queries,serializations,hashes,allocated_bytes,cpu_ms,gc_count,gc_ms,elapsed_ms,edit_p99_ms,fetch_p99_ms,max_publish_lag_ms,final_convergence_ms
+1,1,0,0,0,1000,256,100,100,100,100,218876872,671.875,1,1,128.352,2.396,1.025,3.565,1.093
+1,1,0,10,0,1000,256,100,1,1,1,12077704,93.750,0,0,22.594,0.078,0.727,16.344,16.344
+1,1,0,10,1,1000,256,100,1,1,1,12996336,109.375,0,0,23.958,0.167,0.679,18.078,18.078
+1,1,0,0,1,1000,256,100,100,100,100,216895248,218.750,10,9,117.227,2.397,1.251,3.994,2.042
+1,1,0,0,2,1000,256,100,100,100,100,216641784,125.000,1,1,102.701,1.097,0.768,1.948,0.989
+1,1,0,10,2,1000,256,100,1,1,1,12823128,15.625,0,0,26.813,0.071,0.601,23.166,23.166
+1,1,5,0,0,1000,256,100,100,100,100,239381304,578.125,2,3,504.622,2.046,0.780,4.033,0.987
+1,1,5,10,0,1000,256,100,33,33,33,106512848,250.000,3,2,521.962,0.330,0.853,16.584,1.378
+1,1,5,10,1,1000,256,100,33,33,33,105874160,359.375,4,4,523.376,0.294,0.955,16.555,16.555
+1,1,5,0,1,1000,256,100,100,100,100,238510160,234.375,17,13,501.673,2.873,0.931,2.776,0.955
+1,1,5,0,2,1000,256,100,100,100,100,238484864,1234.375,29,19,497.829,2.489,1.313,2.511,0.967
+1,1,5,10,2,1000,256,100,32,32,32,102562768,109.375,10,7,517.533,0.239,0.747,16.969,12.846
+1,1,25,0,0,1000,256,100,100,100,100,349373552,359.375,1,1,2489.179,1.887,0.912,2.010,1.198
+1,1,25,10,0,1000,256,100,100,100,100,355900336,218.750,1,1,2500.557,0.204,0.946,24.438,17.490
+1,1,25,10,1,1000,256,100,100,100,100,355916552,375.000,2,2,2498.587,0.204,0.903,22.275,17.509
+1,1,25,0,1,1000,256,100,100,100,100,348501552,406.250,1,1,2482.540,1.848,0.839,2.312,1.525
+1,1,25,0,2,1000,256,100,100,100,100,348493088,0.000,1,1,2481.802,1.824,0.947,2.951,1.356
+1,1,25,10,2,1000,256,100,100,100,100,356091848,625.000,2,2,2502.575,0.187,0.844,22.656,17.781
+1,2,0,0,0,1000,256,100,100,100,100,230511552,156.250,1,1,102.723,1.206,0.762,2.279,0.989
+1,2,0,10,0,1000,256,100,2,2,2,9980072,31.250,0,0,15.496,0.031,0.714,14.222,14.222
+1,2,0,10,1,1000,256,100,2,2,2,22508680,0.000,0,0,28.474,0.034,0.735,26.056,26.056
+1,2,0,0,1,1000,256,100,100,100,100,232291232,78.125,2,3,102.641,2.173,1.657,2.399,0.995
+1,2,0,0,2,1000,256,100,100,100,100,230501640,109.375,1,1,98.887,1.123,1.864,2.508,0.964
+1,2,0,10,2,1000,256,100,2,2,2,17137000,0.000,0,0,15.678,0.031,0.733,13.160,13.160
+1,2,5,0,0,1000,256,100,100,100,100,277039424,0.000,1,1,499.142,1.713,0.931,2.280,1.019
+1,2,5,10,0,1000,256,100,49,49,49,178006016,0.000,0,0,513.523,0.214,0.892,17.181,16.323
+1,2,5,10,1,1000,256,100,48,48,48,172429656,15.625,1,1,507.818,0.152,1.676,18.099,11.629
+1,2,5,0,1,1000,256,100,100,100,100,277033024,0.000,1,1,497.660,1.702,0.890,2.259,0.982
+1,2,5,0,2,1000,256,100,100,100,100,278822936,640.625,1,2,507.720,1.874,2.027,3.047,1.341
+1,2,5,10,2,1000,256,100,48,48,48,172231664,125.000,1,1,509.922,0.154,1.205,17.850,12.131
+1,2,25,0,0,1000,256,100,100,100,100,504354424,203.125,2,2,2481.064,2.543,0.961,2.479,1.293
+1,2,25,10,0,1000,256,100,100,100,100,510455760,359.375,2,2,2487.504,0.180,0.978,18.275,17.464
+1,2,25,10,1,1000,256,100,100,100,100,512035768,125.000,1,1,2489.723,0.190,1.031,18.439,17.277
+1,2,25,0,1,1000,256,100,100,100,100,504363448,0.000,1,1,2481.679,2.193,0.950,2.685,1.274
+1,2,25,0,2,1000,256,100,100,100,100,504363448,453.125,1,1,2477.516,1.781,0.841,2.390,1.528
+1,2,25,10,2,1000,256,100,100,100,100,517401472,0.000,1,1,2505.326,0.181,0.810,17.683,16.823
+2,1,0,0,0,1000,256,100,100,100,100,224524608,140.625,1,1,93.229,1.202,0.775,2.148,0.890
+2,1,0,10,0,1000,256,100,2,2,2,18429856,0.000,0,0,18.311,0.020,0.773,16.311,16.311
+2,1,0,10,1,1000,256,100,2,2,2,9798128,31.250,0,0,16.667,0.024,0.811,14.531,14.531
+2,1,0,0,1,1000,256,100,100,100,100,224524920,93.750,2,2,94.845,2.183,1.464,2.206,0.903
+2,1,0,0,2,1000,256,100,100,100,100,224524656,93.750,1,1,95.548,1.331,0.814,2.232,0.949
+2,1,0,10,2,1000,256,100,2,2,2,11495288,0.000,0,0,16.119,0.024,0.580,14.647,14.647
+2,1,5,0,0,1000,256,100,100,100,100,271332472,31.250,2,2,503.334,2.138,1.044,2.176,1.520
+2,1,5,10,0,1000,256,100,48,48,48,169194088,375.000,1,1,522.215,0.111,0.812,16.913,11.710
+2,1,5,10,1,1000,256,100,49,49,49,171297928,62.500,0,0,511.242,0.243,1.083,17.951,12.363
+2,1,5,0,1,1000,256,100,100,100,100,271330840,31.250,2,2,505.416,2.146,1.650,2.728,1.331
+2,1,5,0,2,1000,256,100,100,100,100,271330576,0.000,1,1,505.260,1.624,0.778,2.192,1.202
+2,1,5,10,2,1000,256,100,47,47,47,165359088,0.000,1,1,509.694,0.106,0.817,17.378,11.251
+2,1,25,0,0,1000,256,100,100,100,100,491493512,187.500,2,2,2484.231,2.616,0.904,2.977,1.372
+2,1,25,10,0,1000,256,100,100,100,100,498917104,218.750,1,1,2495.815,0.175,0.974,17.825,17.325
+2,1,25,10,1,1000,256,100,100,100,100,504105536,31.250,1,1,2507.086,0.153,0.818,17.531,16.808
+2,1,25,0,1,1000,256,100,100,100,100,491493344,828.125,1,1,2486.209,1.680,0.834,2.246,1.143
+2,1,25,0,2,1000,256,100,100,100,100,491493248,0.000,1,1,2490.748,1.944,0.953,2.508,1.471
+2,1,25,10,2,1000,256,100,100,100,100,497473920,15.625,2,2,2492.128,0.182,0.944,18.293,16.888
+2,2,0,0,0,1000,256,100,100,100,100,243109856,125.000,2,2,100.004,2.283,1.018,2.299,2.259
+2,2,0,10,0,1000,256,100,4,4,4,32234472,93.750,0,0,17.775,0.025,0.761,15.517,15.517
+2,2,0,10,1,1000,256,100,4,4,4,28695536,31.250,0,0,16.283,0.019,0.750,14.774,14.774
+2,2,0,0,1,1000,256,100,100,100,100,243093344,93.750,1,1,97.313,1.161,0.781,2.265,1.000
+2,2,0,0,2,1000,256,100,100,100,100,243109568,93.750,1,1,98.634,1.145,0.784,2.387,0.977
+2,2,0,10,2,1000,256,100,4,4,4,32235024,15.625,1,2,18.859,0.023,0.748,17.576,17.576
+2,2,5,0,0,1000,256,100,100,100,100,336183320,15.625,1,1,501.563,1.710,0.977,2.922,1.304
+2,2,5,10,0,1000,256,100,73,73,73,293938768,15.625,1,1,513.968,0.149,0.805,19.462,17.468
+2,2,5,10,1,1000,256,100,73,73,73,295391432,171.875,1,2,523.952,0.150,0.766,18.673,18.476
+2,2,5,0,1,1000,256,100,100,100,100,336196120,125.000,1,1,498.489,1.946,0.798,2.338,1.075
+2,2,5,0,2,1000,256,100,100,100,100,336196120,921.875,1,1,497.201,1.723,0.841,2.208,1.275
+2,2,5,10,2,1000,256,100,72,72,72,290604072,328.125,1,1,522.695,0.164,0.899,19.115,16.586
+2,2,25,0,0,1000,256,100,100,100,100,793725520,0.000,2,2,2480.874,1.928,0.796,2.743,1.369
+2,2,25,10,0,1000,256,100,100,100,100,801252368,218.750,2,2,2488.670,0.152,0.824,18.599,16.791
+2,2,25,10,1,1000,256,100,100,100,100,812807960,0.000,2,2,2504.339,0.175,0.819,17.424,17.411
+2,2,25,0,1,1000,256,100,100,100,100,790133416,0.000,3,3,2486.083,2.602,0.803,2.838,1.426
+2,2,25,0,2,1000,256,100,100,100,100,793587384,1343.750,3,3,2486.734,2.821,0.816,2.859,1.440
+2,2,25,10,2,1000,256,100,100,100,100,812821152,15.625,2,2,2508.207,0.186,0.778,17.887,16.998
diff --git a/docs/benchmarks/http-cache-rebuild-4096.csv
b/docs/benchmarks/http-cache-rebuild-4096.csv
new file mode 100644
index 0000000000..d90ad76f9d
--- /dev/null
+++ b/docs/benchmarks/http-cache-rebuild-4096.csv
@@ -0,0 +1,73 @@
+namespaces,groups,pace_ms,window_ms,sample,rows_per_key,handle_chars,events,queries,serializations,hashes,allocated_bytes,cpu_ms,gc_count,gc_ms,elapsed_ms,edit_p99_ms,fetch_p99_ms,max_publish_lag_ms,final_convergence_ms
+1,1,0,0,0,1000,4096,100,100,100,100,2501958776,1859.375,16,15,804.569,9.551,8.286,11.239,7.494
+1,1,0,10,0,1000,4096,100,1,1,1,97111200,156.250,1,1,34.881,0.233,5.426,27.260,27.260
+1,1,0,10,1,1000,4096,100,1,1,1,79683312,156.250,0,0,25.869,0.078,4.879,20.326,20.326
+1,1,0,0,1,1000,4096,100,100,100,100,2508539240,1562.500,33,27,820.990,10.426,8.011,10.220,8.665
+1,1,0,0,2,1000,4096,100,100,100,100,2493433520,828.125,36,29,817.579,9.858,7.506,9.760,8.602
+1,1,0,10,2,1000,4096,100,1,1,1,69945856,31.250,0,0,21.447,0.038,5.187,18.212,18.212
+1,1,5,0,0,1000,4096,100,100,100,100,2508508688,1140.625,44,37,813.621,9.499,7.290,9.473,8.763
+1,1,5,10,0,1000,4096,100,17,17,17,835690472,281.250,8,7,524.470,0.433,6.472,25.018,25.018
+1,1,5,10,1,1000,4096,100,17,17,17,831492600,562.500,8,7,523.678,0.370,7.080,25.013,24.788
+1,1,5,0,1,1000,4096,100,100,100,100,2523590104,1156.250,52,42,832.339,9.711,8.365,9.643,7.721
+1,1,5,0,2,1000,4096,100,100,100,100,2538596496,1765.625,75,58,852.424,9.685,8.426,10.337,8.301
+1,1,5,10,2,1000,4096,100,17,17,17,835778624,1046.875,16,14,521.356,0.189,6.594,24.917,24.670
+1,1,25,0,0,1000,4096,100,100,100,100,4123111864,1156.250,159,121,2488.383,10.788,7.129,10.674,9.495
+1,1,25,10,0,1000,4096,100,81,81,81,3859576840,2921.875,242,162,2515.550,0.196,7.982,26.873,26.345
+1,1,25,10,1,1000,4096,100,81,81,81,3842741520,2921.875,115,73,2504.771,0.214,8.183,34.183,18.425
+1,1,25,0,1,1000,4096,100,100,100,100,4123083656,2484.375,13,13,2484.484,9.102,6.539,9.174,7.407
+1,1,25,0,2,1000,4096,100,100,100,100,4123072624,3312.500,15,13,2483.706,8.908,6.373,8.944,8.648
+1,1,25,10,2,1000,4096,100,81,81,81,3858292920,1234.375,11,10,2510.501,0.232,6.345,38.547,23.052
+1,2,0,0,0,1000,4096,100,100,100,100,3074231024,750.000,15,13,707.330,8.394,6.378,8.420,6.968
+1,2,0,10,0,1000,4096,100,2,2,2,120066512,31.250,1,1,28.667,0.033,5.271,26.457,26.457
+1,2,0,10,1,1000,4096,100,2,2,2,114664296,31.250,0,0,26.410,0.029,4.873,23.282,23.282
+1,2,0,0,1,1000,4096,100,100,100,100,3074226000,1390.625,17,16,712.181,8.748,6.604,8.742,6.778
+1,2,0,0,2,1000,4096,100,100,100,100,3097407272,1578.125,21,20,722.293,9.000,6.819,8.932,8.439
+1,2,0,10,2,1000,4096,100,2,2,2,119963872,31.250,1,1,28.985,0.032,5.461,25.089,25.089
+1,2,5,0,0,1000,4096,100,100,100,100,3177112856,843.750,40,34,748.670,8.681,6.717,9.083,8.602
+1,2,5,10,0,1000,4096,100,35,35,35,1260876816,671.875,17,17,526.902,0.226,9.456,25.399,18.686
+1,2,5,10,1,1000,4096,100,37,37,37,1303940880,703.125,25,25,526.750,0.239,9.169,25.152,19.755
+1,2,5,0,1,1000,4096,100,100,100,100,3340954720,1968.750,112,88,832.952,9.950,7.511,9.989,8.843
+1,2,5,0,2,1000,4096,100,100,100,100,2575986216,1812.500,53,45,881.170,16.536,21.235,25.022,8.696
+1,2,5,10,2,1000,4096,100,34,34,34,1142108120,593.750,6,6,519.258,0.212,25.105,32.335,20.101
+1,2,25,0,0,1000,4096,100,100,100,100,5432644640,5062.500,61,52,2491.964,17.599,19.831,17.588,16.821
+1,2,25,10,0,1000,4096,100,100,100,100,4847604352,6671.875,80,75,2517.146,0.236,20.472,33.401,33.166
+1,2,25,10,1,1000,4096,100,100,100,100,4708704824,6343.750,80,73,2504.799,0.243,20.806,34.338,33.016
+1,2,25,0,1,1000,4096,100,100,100,100,4597079952,5468.750,88,78,2505.398,17.781,20.706,19.557,17.008
+1,2,25,0,2,1000,4096,100,100,100,100,4628366056,8843.750,90,80,2502.475,17.983,21.155,26.029,16.919
+1,2,25,10,2,1000,4096,100,100,100,100,4587705184,7359.375,91,79,2515.831,0.318,21.052,33.251,32.522
+2,1,0,0,0,1000,4096,100,100,100,100,2686624608,1578.125,13,12,836.439,17.411,22.879,22.412,8.360
+2,1,0,10,0,1000,4096,100,2,2,2,113740208,62.500,2,2,43.351,0.030,7.636,41.079,41.079
+2,1,0,10,1,1000,4096,100,2,2,2,124828568,171.875,1,1,44.137,0.036,7.398,41.047,41.047
+2,1,0,0,1,1000,4096,100,100,100,100,3082208752,906.250,17,13,760.568,21.277,21.250,26.487,6.979
+2,1,0,0,2,1000,4096,100,100,100,100,2992198088,2453.125,36,28,829.919,21.404,21.551,21.489,9.162
+2,1,0,10,2,1000,4096,100,2,2,2,102555848,234.375,3,3,41.728,0.025,16.407,40.669,40.669
+2,1,5,0,0,1000,4096,100,100,100,100,3057772784,4031.250,79,59,1205.560,23.879,22.103,25.005,15.538
+2,1,5,10,0,1000,4096,100,34,34,34,1546954632,484.375,9,7,547.442,0.236,18.132,38.734,38.734
+2,1,5,10,1,1000,4096,100,34,34,34,1608904496,406.250,7,6,537.946,0.205,17.383,38.606,38.606
+2,1,5,0,1,1000,4096,100,100,100,100,3073613112,1734.375,17,16,761.704,18.222,19.770,21.009,9.139
+2,1,5,0,2,1000,4096,100,100,100,100,3066909520,1500.000,15,14,735.889,19.492,18.552,24.198,6.950
+2,1,5,10,2,1000,4096,100,34,34,34,1577078648,765.625,8,8,537.649,0.188,6.456,32.873,29.730
+2,1,25,0,0,1000,4096,100,100,100,100,5541799728,4421.875,56,46,2487.316,19.552,17.826,21.165,16.486
+2,1,25,10,0,1000,4096,100,100,100,100,5429957568,3703.125,59,47,2508.953,0.301,18.020,33.242,31.384
+2,1,25,10,1,1000,4096,100,100,100,100,6420931288,3734.375,31,25,2508.530,0.212,15.769,33.107,23.444
+2,1,25,0,1,1000,4096,100,100,100,100,5981170728,2937.500,35,28,2499.376,17.301,17.769,17.194,16.722
+2,1,25,0,2,1000,4096,100,100,100,100,6531360496,4390.625,33,30,2481.787,9.291,6.588,9.295,7.726
+2,1,25,10,2,1000,4096,100,100,100,100,6558445536,1562.500,33,31,2496.811,0.235,6.444,25.469,24.479
+2,2,0,0,0,1000,4096,100,100,100,100,3104555640,1250.000,14,13,777.860,21.895,32.623,23.241,6.996
+2,2,0,10,0,1000,4096,100,4,4,4,234955144,78.125,2,1,55.898,0.025,5.867,53.309,53.309
+2,2,0,10,1,1000,4096,100,4,4,4,191588176,187.500,1,1,51.097,0.024,14.073,48.245,48.245
+2,2,0,0,1,1000,4096,100,100,100,100,3142363880,1718.750,14,12,773.177,16.131,20.018,18.318,7.266
+2,2,0,0,2,1000,4096,100,100,100,100,3104555376,1234.375,16,14,736.585,13.983,20.821,14.397,7.336
+2,2,0,10,2,1000,4096,100,4,4,4,208775160,78.125,2,2,48.395,0.023,5.658,46.600,46.600
+2,2,5,0,0,1000,4096,100,100,100,100,3128486744,1171.875,16,13,729.063,11.681,7.588,13.905,13.905
+2,2,5,10,0,1000,4096,100,60,60,60,2016881504,875.000,10,10,530.052,0.209,21.906,36.417,34.186
+2,2,5,10,1,1000,4096,100,64,64,64,2107876856,1031.250,10,8,527.499,0.279,16.616,32.864,28.490
+2,2,5,0,1,1000,4096,100,100,100,100,3117810240,1218.750,14,12,731.658,16.384,17.435,17.143,7.805
+2,2,5,0,2,1000,4096,100,100,100,100,3267494888,1296.875,69,55,786.463,9.523,7.019,9.988,8.101
+2,2,5,10,2,1000,4096,100,66,66,66,2174786200,1140.625,10,10,527.627,0.186,17.584,34.499,25.696
+2,2,25,0,0,1000,4096,100,100,100,100,6457537944,2890.625,30,25,2487.921,17.217,15.127,17.277,8.953
+2,2,25,10,0,1000,4096,100,100,100,100,6494157912,2968.750,33,27,2512.869,0.293,16.400,32.895,23.669
+2,2,25,10,1,1000,4096,100,100,100,100,6472350952,2531.250,33,26,2497.611,0.236,15.364,33.274,23.510
+2,2,25,0,1,1000,4096,100,100,100,100,6517576816,1921.875,35,27,2495.031,17.628,15.166,17.507,7.779
+2,2,25,0,2,1000,4096,100,100,100,100,5555343384,5796.875,71,59,2494.199,17.943,26.007,23.223,17.283
+2,2,25,10,2,1000,4096,100,100,100,100,5796829944,8937.500,61,47,2508.890,0.331,25.705,33.509,32.356
diff --git a/docs/http-cache-rebuild-coalescing.md
b/docs/http-cache-rebuild-coalescing.md
new file mode 100644
index 0000000000..f4e0aaf8a8
--- /dev/null
+++ b/docs/http-cache-rebuild-coalescing.md
@@ -0,0 +1,222 @@
+# HTTP cache rebuild coalescing
+
+This implements the producer-side experiment in
[#7839](https://github.com/apache/shenyu/issues/7839).
+It applies to independent committed configuration events handled by the HTTP
cache listener.
+Selector import batching and the event dispatcher are unchanged.
+
+## Configuration and compatibility
+
+```yaml
+shenyu:
+ sync:
+ http:
+ enabled: true
+ cache-rebuild-window: 10ms
+ cache-rebuild-max-pending-keys: 1024
+ cache-rebuild-threads: 1
+```
+
+The default window is **0ms**, preserving synchronous event processing. Enable
a positive
+window only when repeated updates to the same namespace/group justify the
latency cost.
+Accepted windows are zero or 1ms–1s; the pending-key capacity must be
positive; the
+worker thread count must be between 1 and 64 (default 1).
+Configuration takes effect when the listener is constructed; restart after
changing it.
+Return the window to zero and restart to restore the original event path.
+
+The existing seven event-driven groups use the hook: app auth, plugin, rule,
selector,
+metadata, proxy selector and discovery upstream. Full startup/periodic
refreshes,
+including the existing AI proxy API key cache path, retain their existing
scope.
+
+## Scheduling and correctness
+
+- Each namespace/group owns at most one queued or running rebuild. Its first
event sets
+ a fixed deadline. Later events increment a generation without resetting that
deadline.
+- A rebuild reads the authoritative group from the database, then uses the
existing Gson,
+ MD5 and version logic. An empty database result publishes an authoritative
empty group.
+ Identical JSON preserves the previous cache object and version.
+- A generation change during a rebuild schedules another pass. Comparing the
generation
+ and removing the pending marker share the same monitor as event admission,
preventing
+ a lost wakeup at the end of a pass.
+- Only snapshot work is combined. Every event still reaches the dispatcher,
other
+ transport listeners and per-event callbacks. DELETE/UPDATE payloads are
never merged
+ into a synthetic event. Existing after-commit delivery is retained; rollback
cannot
+ enqueue a rebuild. Event payloads are not retained in the queue.
+- Successful publication queues the existing long-poll notification. When
enabled,
+ long-poll registration also rechecks MD5/version on the notification
scheduler to close
+ the compare/register race.
+- Namespace locks serialize snapshot loads/publications with periodic and
client-triggered
+ repair, preventing an older concurrent read from overwriting a newer
publication.
+ No event triggers a global refresh.
+- A separate pool of `cache-rebuild-threads` workers performs rebuilds; the
existing
+ notification/timer scheduler keeps its existing thread count. Each key has
at most one
+ scheduled or running drain, so several workers rebuild distinct keys in
parallel but never
+ the same key concurrently. At capacity, a new key rebuilds synchronously on
its committing
+ caller. Existing pending keys continue to coalesce. No unbounded per-event
task list is created.
+- A failed rebuild retains its dirty generation and retries with exponential
backoff: 1s, 2s,
+ 4s, … capped at 30s, never shorter than the window. The count resets after a
successful
+ rebuild. Retries never give up, because dropping the key would leave
long-poll readers on
+ a stale snapshot; backoff keeps a permanently failing key to at most one
error log every 30s.
+ The key yields to other keys between attempts, and the last successfully
published cache
+ remains readable.
+- Shutdown cancels queued work and interrupts the workers. A commit that races
shutdown is
+ rebuilt synchronously on its caller instead of failing its after-commit
hook; the
+ long-poll notification is skipped once the notification scheduler has
stopped.
+
+### Worker threads trade-off
+
+With the default single worker, coalescing trades burst redundancy for
serialization:
+without coalescing, distinct namespaces rebuild in parallel on their
committing threads,
+while with one worker every key queues behind the same thread. Because a
rebuild also holds
+its namespace lock, one slow namespace delays unrelated groups queued behind
it. The recorded
+benchmark used one worker and at most four active keys. Deployments where many
namespaces or
+groups change concurrently should raise `cache-rebuild-threads` (for example
to the number of
+actively edited namespaces, bounded by database connection capacity). Rebuilds
of different
+groups in the same namespace still serialize on that namespace's lock, as they
do today.
+
+For a healthy idle key, the **intentional added delay is at most the
configured window W**.
+Publication also needs queue wait Q and rebuild time R, followed by
notification scheduling.
+An event arriving during a rebuild can additionally wait for that in-flight
read to finish.
+With at most K pending keys and bounded rebuild time Rmax, one queue pass
costs at most
+approximately K × Rmax; a slow database, GC pause or blocked notification
scheduler can
+exceed a wall-clock target. This is not a promise of 10ms end-to-end delivery
under overload.
+The local benchmark below gates final cache publication at **250ms** from the
last committed
+version of every key. Production rollout needs an explicit end-to-end
convergence budget.
+
+## Reproduce the tests
+
+Use Java 17 and the Maven wrapper (on Windows, `./mvnw.cmd`).
+
+```sh
+./mvnw -B -ntp -pl shenyu-admin -am test \
+
-Dtest=HttpCacheRebuildCoordinatorTest,HttpCacheRebuildTest,AbstractDataChangedListenerTest,HttpLongPollingDataChangedListenerTest,HttpSyncPropertiesTest,DataChangedEventDispatcherTest,SelectorServiceTest
\
+ -Dsurefire.failIfNoSpecifiedTests=false -Djacoco.skip=true \
+ -DskipRemoteResources=true -Dmaven.javadoc.skip=true
+./mvnw -B -ntp -pl shenyu-admin apache-rat:check
+```
+
+Tests cover concurrent independent H2 commits, rollback, event/callback
preservation,
+DELETE followed by UPDATE, authoritative empty groups, namespace/group
separation,
+capacity fallback, unchanged MD5/version, disabled mode, notification ordering
and the
+registration race. Deterministic coordinator tests cover in-flight writes,
retry without
+another event, capped exponential backoff and its reset, fixed deadlines, key
reuse, rejected
+scheduling and shutdown (including a commit racing shutdown). Multi-worker
tests show that
+distinct keys rebuild in parallel while one key never rebuilds concurrently.
+The existing selector import tests run unchanged.
+
+PR verification on master `eb0c32ff7` plus this change: **105 tests passed**,
with no failures or errors; reactor Checkstyle
+passed. The admin RAT check approved all 1037 scanned resources, including the
new Java files.
+The benchmark below was recorded before worker threads and failure backoff
became configurable; it ran with
+the default single worker and no failures, so those changes do not alter the
recorded paths.
+
+## Reproduce the producer benchmark
+
+```sh
+./mvnw -B -ntp -pl shenyu-admin -am test \
+ -Dtest=HttpCacheRebuildBenchmark -Dsurefire.failIfNoSpecifiedTests=false \
+ -Dshenyu.http.benchmark=true -DargLine="-Xms512m -Xmx512m" \
+ -Djacoco.skip=true -DskipRemoteResources=true -Dmaven.javadoc.skip=true
+```
+
+The output is `shenyu-admin/target/http-cache-rebuild-benchmark.csv`. Optional
properties:
+`shenyu.http.benchmark.rows` (1000), `events` (100), `handleChars` (256),
+`warmup` (5), `samples` (3), and `output` (the CSV path); all use the same
+`shenyu.http.benchmark.` prefix. Set `handleChars=4096` to repeat with larger
handles.
+
+Each variant uses real H2 commits and queries, the real event dispatcher, the
HTTP listener,
+and complete DTO serialization/hashing. A small JDBC-backed service fixture
substitutes for
+MyBatis; SQL counts are actual fixture SELECTs, not estimates of production
mapper query
+fan-out. RULE and SELECTOR have 1000 rows per key, with synthetic handles and
no conditions.
+The matrix covers one/two namespaces × one/two groups, with burst, 5ms and
25ms spacing.
+Every update is a separate transaction, never an import. A serial producer
follows a fixed
+arrival schedule; if it falls behind, it catches up. Burst mode runs as fast
as the caller
+can commit, so elapsed time is part of the comparison, not a fixed-rate
saturation claim.
+
+Each trial warms the write/rebuild and fetch paths, alternates variant order
across three samples,
+and runs a concurrent in-process `fetchConfig` reader. CSV metrics include
full SELECT,
+serialization and hash counts, live-JVM-thread allocated bytes, JVM process
CPU time, GC count/time,
+edit p99, fetch p99, lag for published versions and final convergence.
Allocation totals
+include the reader and JVM background threads alive at sampling; process CPU
includes
+all JVM threads (Windows per-thread CPU deltas are too coarse here). These are
not isolated
+production allocation profiles. The fetch path is measured in-process, without
HTTP,
+gateway plugins, a network or business traffic. Intermediate overwritten
versions are
+intentionally not required to be published.
+
+## Recorded local results
+
+Measured on 2026-10-19 with Microsoft OpenJDK 17.0.18+8-LTS, Windows 11
+(10.0.26200), AMD Ryzen 9 8945HX, 32 logical processors, approximately 32 GiB
RAM,
+and a fixed 512 MiB JVM heap. Base checkout:
`297ac6f849857bd0c2ad368709013e98494b0330`
+plus this change. Each payload size ran in a separate JVM with three samples
per scenario,
+five warmup traces of 100 independent commits, and 20 fetch warmups per key.
+
+Both runs used the command above with an explicit output path. The second also
set
+`-Dshenyu.http.benchmark.handleChars=4096`. All other parameters were
identical.
+
+Raw results: [256-character handles](benchmarks/http-cache-rebuild-256.csv) and
+[4096-character handles](benchmarks/http-cache-rebuild-4096.csv), 72 measured
trials each.
+Each table entry is the median of three trials, except the final column, which
is the
+maximum final convergence across those trials. Pairs show **disabled → 10ms
window**.
+Every full rebuild below is one actual fixture SELECT, one full serialization
and one hash.
+Allocation totals include concurrent reads and depend on trace completion time.
+CPU and GC observations are retained in the CSV. Windows CPU counters remain
coarsely
+quantized (including zero deltas); do not interpret those zero values as zero
CPU use or
+derive a CPU-speedup claim from this run. No gateway business p99 is
represented here.
+
+### 256-character handles
+
+| Namespaces × groups | Spacing | Full rebuilds | Allocated MiB | Edit p99 ms
| Fetch p99 ms | Max final convergence, enabled (ms) |
+| --- | --- | --- | --- | --- | --- | --- |
+| 1 × 1 | burst | 100 → 1 | 206.8 → 12.2 | 2.396 → 0.078 | 1.025 → 0.679 |
23.166 |
+| 1 × 1 | 5ms | 100 → 33 | 227.5 → 101.0 | 2.489 → 0.294 | 0.931 → 0.853 |
16.555 |
+| 1 × 1 | 25ms | 100 → 100 | 332.4 → 339.4 | 1.848 → 0.204 | 0.912 → 0.903 |
17.781 |
+| 1 × 2 | burst | 100 → 2 | 219.8 → 16.3 | 1.206 → 0.031 | 1.657 → 0.733 |
26.056 |
+| 1 × 2 | 5ms | 100 → 48 | 264.2 → 164.4 | 1.713 → 0.154 | 0.931 → 1.205 |
16.323 |
+| 1 × 2 | 25ms | 100 → 100 | 481.0 → 488.3 | 2.193 → 0.181 | 0.950 → 0.978 |
17.464 |
+| 2 × 1 | burst | 100 → 2 | 214.1 → 11.0 | 1.331 → 0.024 | 0.814 → 0.773 |
16.311 |
+| 2 × 1 | 5ms | 100 → 48 | 258.8 → 161.4 | 2.138 → 0.111 | 1.044 → 0.817 |
12.363 |
+| 2 × 1 | 25ms | 100 → 100 | 468.7 → 475.8 | 1.944 → 0.175 | 0.904 → 0.944 |
17.325 |
+| 2 × 2 | burst | 100 → 4 | 231.8 → 30.7 | 1.161 → 0.023 | 0.784 → 0.750 |
17.576 |
+| 2 × 2 | 5ms | 100 → 73 | 320.6 → 280.3 | 1.723 → 0.150 | 0.841 → 0.805 |
18.476 |
+| 2 × 2 | 25ms | 100 → 100 | 756.8 → 775.2 | 2.602 → 0.175 | 0.803 → 0.819 |
17.411 |
+
+### 4096-character handles
+
+| Namespaces × groups | Spacing | Full rebuilds | Allocated MiB | Edit p99 ms
| Fetch p99 ms | Max final convergence, enabled (ms) |
+| --- | --- | --- | --- | --- | --- | --- |
+| 1 × 1 | burst | 100 → 1 | 2386.1 → 76.0 | 9.858 → 0.078 | 8.011 → 5.187 |
27.260 |
+| 1 × 1 | 5ms | 100 → 17 | 2406.7 → 797.0 | 9.685 → 0.370 | 8.365 → 6.594 |
25.018 |
+| 1 × 1 | 25ms | 100 → 81 | 3932.1 → 3679.6 | 9.102 → 0.214 | 6.539 → 7.982 |
26.345 |
+| 1 × 2 | burst | 100 → 2 | 2931.8 → 114.4 | 8.748 → 0.032 | 6.604 → 5.271 |
26.457 |
+| 1 × 2 | 5ms | 100 → 35 | 3029.9 → 1202.5 | 9.950 → 0.226 | 7.511 → 9.456 |
20.101 |
+| 1 × 2 | 25ms | 100 → 100 | 4414.0 → 4490.6 | 17.781 → 0.243 | 20.706 →
20.806 | 33.166 |
+| 2 × 1 | burst | 100 → 2 | 2853.6 → 108.5 | 21.277 → 0.030 | 21.551 → 7.636 |
41.079 |
+| 2 × 1 | 5ms | 100 → 34 | 2924.8 → 1504.0 | 19.492 → 0.205 | 19.770 → 17.383
| 38.734 |
+| 2 × 1 | 25ms | 100 → 100 | 5704.1 → 6123.5 | 17.301 → 0.235 | 17.769 →
15.769 | 31.384 |
+| 2 × 2 | burst | 100 → 4 | 2960.7 → 199.1 | 16.131 → 0.024 | 20.821 → 5.867 |
53.309 |
+| 2 × 2 | 5ms | 100 → 64 | 2983.6 → 2010.2 | 11.681 → 0.209 | 7.588 → 17.584 |
34.186 |
+| 2 × 2 | 25ms | 100 → 100 | 6158.4 → 6172.5 | 17.628 → 0.293 | 15.166 →
16.400 | 32.356 |
+
+Across all measured burst scenarios, 100 committed events required **1–4 full
rebuilds**,
+instead of 100. The largest enabled final convergence was **53.309ms**, below
the 250ms
+local publication budget. With 256-character handles, 25ms-spaced updates
performed
+100 rebuilds in both modes and slightly increased allocation when enabled: use
zero for
+that workload. With larger handles, some 25ms traces still overlapped a
rebuild and merged;
+this does not establish a universal benefit for sparse traffic.
+
+These measurements support an opt-in window for repeated hot-key updates. They
do not
+support changing the default or promising lower deployed business-request p99.
+
+## Deployment validation boundary
+
+These producer results do **not** establish deployed gateway business-request
p99 or
+network convergence. The local Docker engine was unavailable during this run,
so the
+Docker HTTP integration suite and an admin + gateway + backend load run were
not executed.
+The default therefore remains disabled.
+
+Before enabling in a deployment, repeat baseline/window runs with identical
database,
+admin, gateway and backend resources, warm all request paths, repeat the same
committed
+update trace, and measure database statements, CPU/allocation/GC, gateway
convergence,
+business-request p99 and errors. Require a query/allocation benefit larger
than sample
+variation while meeting the chosen convergence budget; leave the window at
zero for
+sparse updates. See the repository's [integration test
setup](../shenyu-integrated-test/README.md)
+and [benchmark contract](https://github.com/apache/shenyu/issues/7650).
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/properties/HttpSyncProperties.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/properties/HttpSyncProperties.java
index 0e751910b6..f8909381ac 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/properties/HttpSyncProperties.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/config/properties/HttpSyncProperties.java
@@ -20,6 +20,7 @@ package org.apache.shenyu.admin.config.properties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import java.time.Duration;
+import java.util.Objects;
/**
* the http sync strategy properties.
@@ -42,6 +43,87 @@ public class HttpSyncProperties {
*/
private int notifyBatchSize = 100;
+ /**
+ * Fixed cache rebuild coalescing window. Zero preserves synchronous event
processing.
+ */
+ private Duration cacheRebuildWindow = Duration.ZERO;
+
+ /**
+ * Maximum pending namespace/group keys. New keys rebuild on the caller
when full.
+ */
+ private int cacheRebuildMaxPendingKeys = 1024;
+
+ /**
+ * Worker threads rebuilding coalesced keys. Distinct namespace/group keys
rebuild in parallel;
+ * one key never rebuilds concurrently.
+ */
+ private int cacheRebuildThreads = 1;
+
+ /**
+ * Get the fixed coalescing window.
+ *
+ * @return cache rebuild window
+ */
+ public Duration getCacheRebuildWindow() {
+ return cacheRebuildWindow;
+ }
+
+ /**
+ * Set the fixed coalescing window.
+ *
+ * @param cacheRebuildWindow zero to disable, or a window from 1 to 1000
milliseconds
+ */
+ public void setCacheRebuildWindow(final Duration cacheRebuildWindow) {
+ if (Objects.isNull(cacheRebuildWindow) ||
cacheRebuildWindow.isNegative()
+ || cacheRebuildWindow.compareTo(Duration.ofSeconds(1)) > 0
+ || (!cacheRebuildWindow.isZero() &&
cacheRebuildWindow.toMillis() == 0)) {
+ throw new IllegalArgumentException("cacheRebuildWindow must be
zero or between 1ms and 1s");
+ }
+ this.cacheRebuildWindow = cacheRebuildWindow;
+ }
+
+ /**
+ * Get the pending key capacity.
+ *
+ * @return maximum pending keys
+ */
+ public int getCacheRebuildMaxPendingKeys() {
+ return cacheRebuildMaxPendingKeys;
+ }
+
+ /**
+ * Set the pending key capacity.
+ *
+ * @param cacheRebuildMaxPendingKeys positive maximum pending keys
+ */
+ public void setCacheRebuildMaxPendingKeys(final int
cacheRebuildMaxPendingKeys) {
+ if (cacheRebuildMaxPendingKeys <= 0) {
+ throw new IllegalArgumentException("cacheRebuildMaxPendingKeys
must be positive");
+ }
+ this.cacheRebuildMaxPendingKeys = cacheRebuildMaxPendingKeys;
+ }
+
+ /**
+ * Get the number of cache rebuild worker threads.
+ *
+ * @return cache rebuild worker threads
+ */
+ public int getCacheRebuildThreads() {
+ return cacheRebuildThreads;
+ }
+
+ /**
+ * Set the number of cache rebuild worker threads.
+ *
+ * @param cacheRebuildThreads worker threads from 1 to 64
+ */
+ public void setCacheRebuildThreads(final int cacheRebuildThreads) {
+ if (cacheRebuildThreads <= 0 || cacheRebuildThreads > 64) {
+ throw new IllegalArgumentException("cacheRebuildThreads must be
between 1 and 64");
+ }
+ this.cacheRebuildThreads = cacheRebuildThreads;
+ }
+
/**
* Gets the value of enabled.
*
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/AbstractDataChangedListener.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/AbstractDataChangedListener.java
index 5415c22553..9f5f991fef 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/AbstractDataChangedListener.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/AbstractDataChangedListener.java
@@ -153,7 +153,7 @@ public abstract class AbstractDataChangedListener
implements DataChangedListener
return;
}
groupByNamespace(changed, AppAuthData::getNamespaceId,
ConfigGroupEnum.APP_AUTH).forEach((namespaceId, namespaceChanged) -> {
- this.updateAppAuthCache(namespaceId);
+ this.refreshCache(ConfigGroupEnum.APP_AUTH, namespaceId, () ->
this.updateAppAuthCache(namespaceId));
this.afterAppAuthChanged(namespaceChanged, eventType, namespaceId);
});
}
@@ -173,7 +173,7 @@ public abstract class AbstractDataChangedListener
implements DataChangedListener
return;
}
groupByNamespace(changed, MetaData::getNamespaceId,
ConfigGroupEnum.META_DATA).forEach((namespaceId, namespaceChanged) -> {
- this.updateMetaDataCache(namespaceId);
+ this.refreshCache(ConfigGroupEnum.META_DATA, namespaceId, () ->
this.updateMetaDataCache(namespaceId));
this.afterMetaDataChanged(namespaceChanged, eventType,
namespaceId);
});
}
@@ -193,7 +193,7 @@ public abstract class AbstractDataChangedListener
implements DataChangedListener
return;
}
groupByNamespace(changed, PluginData::getNamespaceId,
ConfigGroupEnum.PLUGIN).forEach((namespaceId, namespaceChanged) -> {
- this.updatePluginCache(namespaceId);
+ this.refreshCache(ConfigGroupEnum.PLUGIN, namespaceId, () ->
this.updatePluginCache(namespaceId));
this.afterPluginChanged(namespaceChanged, eventType, namespaceId);
});
}
@@ -223,7 +223,7 @@ public abstract class AbstractDataChangedListener
implements DataChangedListener
return;
}
groupByNamespace(changed, RuleData::getNamespaceId,
ConfigGroupEnum.RULE).forEach((namespaceId, namespaceChanged) -> {
- this.updateRuleCache(namespaceId);
+ this.refreshCache(ConfigGroupEnum.RULE, namespaceId, () ->
this.updateRuleCache(namespaceId));
this.afterRuleChanged(namespaceChanged, eventType, namespaceId);
});
}
@@ -243,7 +243,7 @@ public abstract class AbstractDataChangedListener
implements DataChangedListener
return;
}
groupByNamespace(changed, SelectorData::getNamespaceId,
ConfigGroupEnum.SELECTOR).forEach((namespaceId, namespaceChanged) -> {
- this.updateSelectorCache(namespaceId);
+ this.refreshCache(ConfigGroupEnum.SELECTOR, namespaceId, () ->
this.updateSelectorCache(namespaceId));
this.afterSelectorChanged(namespaceChanged, eventType,
namespaceId);
});
}
@@ -260,7 +260,7 @@ public abstract class AbstractDataChangedListener
implements DataChangedListener
return;
}
groupByNamespace(changed, ProxySelectorData::getNamespaceId,
ConfigGroupEnum.PROXY_SELECTOR).forEach((namespaceId, namespaceChanged) -> {
- this.updateProxySelectorDataCache(namespaceId);
+ this.refreshCache(ConfigGroupEnum.PROXY_SELECTOR, namespaceId, ()
-> this.updateProxySelectorDataCache(namespaceId));
this.afterProxySelectorChanged(namespaceChanged, eventType,
namespaceId);
});
}
@@ -286,7 +286,7 @@ public abstract class AbstractDataChangedListener
implements DataChangedListener
return;
}
groupByNamespace(changed, DiscoverySyncData::getNamespaceId,
ConfigGroupEnum.DISCOVER_UPSTREAM).forEach((namespaceId, namespaceChanged) -> {
- this.updateDiscoveryUpstreamDataCache(namespaceId);
+ this.refreshCache(ConfigGroupEnum.DISCOVER_UPSTREAM, namespaceId,
() -> this.updateDiscoveryUpstreamDataCache(namespaceId));
this.afterDiscoveryUpstreamDataChanged(namespaceChanged,
eventType, namespaceId);
});
}
@@ -330,6 +330,18 @@ public abstract class AbstractDataChangedListener
implements DataChangedListener
protected abstract void afterInitialize();
+ /**
+ * Refresh a group for a committed change. Transport implementations may
coalesce snapshot work;
+ * event callbacks are still invoked once for every event and namespace.
+ *
+ * @param group configuration group
+ * @param namespaceId namespace id
+ * @param rebuild authoritative cache rebuild, without event payloads
+ */
+ protected void refreshCache(final ConfigGroupEnum group, final String
namespaceId, final Runnable rebuild) {
+ rebuild.run();
+ }
+
/**
* if md5 is not the same as the original, then update local cache.
*
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/http/HttpCacheRebuildCoordinator.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/http/HttpCacheRebuildCoordinator.java
new file mode 100644
index 0000000000..648b7ca8d4
--- /dev/null
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/http/HttpCacheRebuildCoordinator.java
@@ -0,0 +1,173 @@
+/*
+ * 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.shenyu.admin.listener.http;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Bounded snapshot work, with one scheduled or running rebuild per
namespace/group.
+ * The first event sets the deadline; later events never postpone it.
+ *
+ * <p>Each key has at most one scheduled or running drain, so the executor may
use several
+ * threads: distinct keys rebuild in parallel while a single key is never
rebuilt concurrently.
+ * A failing rebuild keeps its key and retries with exponential backoff,
starting at
+ * {@value #FAILURE_RETRY_MILLIS}ms and capped at {@value
#MAX_FAILURE_RETRY_MILLIS}ms, until a
+ * successful rebuild covers every arrival; it never gives up, because giving
up would leave
+ * long-poll readers on a stale snapshot.</p>
+ */
+final class HttpCacheRebuildCoordinator implements AutoCloseable {
+
+ static final long FAILURE_RETRY_MILLIS = 1000;
+
+ static final long MAX_FAILURE_RETRY_MILLIS = 30_000;
+
+ private static final Logger LOG =
LoggerFactory.getLogger(HttpCacheRebuildCoordinator.class);
+
+ private final Map<String, PendingRebuild> pending = new HashMap<>();
+
+ private final ScheduledExecutorService executor;
+
+ private final long windowMillis;
+
+ private final int maxPendingKeys;
+
+ private boolean closed;
+
+ HttpCacheRebuildCoordinator(final ScheduledExecutorService executor, final
long windowMillis, final int maxPendingKeys) {
+ this.executor = executor;
+ this.windowMillis = windowMillis;
+ this.maxPendingKeys = maxPendingKeys;
+ }
+
+ /**
+ * Mark a committed key dirty without retaining the event payload.
+ *
+ * @param key namespace/group key
+ * @param rebuild reload, publish and notify action
+ * @return false when the caller must rebuild synchronously (capacity,
executor rejection or
+ * a coordinator that is already closed during shutdown)
+ */
+ boolean request(final String key, final Runnable rebuild) {
+ synchronized (pending) {
+ if (closed) {
+ // A commit racing shutdown must not fail its after-commit
hook.
+ return false;
+ }
+ PendingRebuild existing = pending.get(key);
+ if (Objects.nonNull(existing)) {
+ existing.generation++;
+ return true;
+ }
+ if (pending.size() >= maxPendingKeys) {
+ return false;
+ }
+ PendingRebuild next = new PendingRebuild(rebuild);
+ pending.put(key, next);
+ return schedule(key, next, windowMillis);
+ }
+ }
+
+ private boolean schedule(final String key, final PendingRebuild state,
final long delayMillis) {
+ try {
+ executor.schedule(() -> drain(key, state), delayMillis,
TimeUnit.MILLISECONDS);
+ return true;
+ } catch (RejectedExecutionException ex) {
+ pending.remove(key, state);
+ LOG.warn("HTTP cache rebuild scheduling rejected for {}", key, ex);
+ return false;
+ }
+ }
+
+ private void drain(final String key, final PendingRebuild state) {
+ long generation;
+ synchronized (pending) {
+ if (closed) {
+ return;
+ }
+ generation = state.generation;
+ }
+ boolean successful = false;
+ try {
+ state.rebuild.run();
+ successful = true;
+ } catch (Exception ex) {
+ synchronized (pending) {
+ state.failures++;
+ }
+ LOG.error("HTTP cache rebuild failed for {} (attempt {}), will
retry in {}ms",
+ key, state.failures, retryDelayMillis(state.failures), ex);
+ } finally {
+ synchronized (pending) {
+ if (!closed) {
+ if (successful) {
+ state.failures = 0;
+ }
+ if (successful && generation == state.generation) {
+ pending.remove(key, state);
+ } else {
+ // Keep ownership until a successful rebuild covers
all arrivals. Yield between
+ // generations so a hot or failing key cannot
monopolize a worker.
+ schedule(key, state, successful ? windowMillis :
retryDelayMillis(state.failures));
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * Exponential backoff for consecutive failures of one key, never shorter
than the window.
+ *
+ * @param failures consecutive failures, at least one
+ * @return the retry delay in milliseconds
+ */
+ long retryDelayMillis(final int failures) {
+ int doublings = Math.min(Math.max(failures, 1) - 1, 30);
+ long backoff = Math.min(MAX_FAILURE_RETRY_MILLIS, FAILURE_RETRY_MILLIS
<< doublings);
+ return Math.max(windowMillis, backoff);
+ }
+
+ @Override
+ public void close() {
+ synchronized (pending) {
+ closed = true;
+ pending.clear();
+ executor.shutdownNow();
+ }
+ }
+
+ private static final class PendingRebuild {
+
+ private final Runnable rebuild;
+
+ private long generation;
+
+ private int failures;
+
+ private PendingRebuild(final Runnable rebuild) {
+ this.rebuild = rebuild;
+ }
+ }
+}
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/http/HttpLongPollingDataChangedListener.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/http/HttpLongPollingDataChangedListener.java
index f9fcd5976a..95084fb167 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/http/HttpLongPollingDataChangedListener.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/http/HttpLongPollingDataChangedListener.java
@@ -45,6 +45,7 @@ import org.apache.shenyu.common.exception.ShenyuException;
import org.apache.shenyu.common.utils.GsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.DisposableBean;
import org.springframework.http.MediaType;
import jakarta.servlet.AsyncContext;
@@ -62,6 +63,7 @@ import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -77,7 +79,7 @@ import static
org.apache.shenyu.common.constant.Constants.SYS_DEFAULT_NAMESPACE_
* @since 2.0.0
*/
@SuppressWarnings("all")
-public class HttpLongPollingDataChangedListener extends
AbstractDataChangedListener {
+public class HttpLongPollingDataChangedListener extends
AbstractDataChangedListener implements DisposableBean {
private static final Logger LOG =
LoggerFactory.getLogger(HttpLongPollingDataChangedListener.class);
@@ -100,6 +102,8 @@ public class HttpLongPollingDataChangedListener extends
AbstractDataChangedListe
private final HttpSyncProperties httpSyncProperties;
+ private final HttpCacheRebuildCoordinator cacheRebuildCoordinator;
+
/**
* Instantiates a new Http long polling data changed listener.
*
@@ -111,6 +115,60 @@ public class HttpLongPollingDataChangedListener extends
AbstractDataChangedListe
this.scheduler = new ScheduledThreadPoolExecutor(1,
ShenyuThreadFactory.create("long-polling", true));
this.httpSyncProperties = httpSyncProperties;
+ this.cacheRebuildCoordinator =
Objects.nonNull(httpSyncProperties.getCacheRebuildWindow()) &&
!httpSyncProperties.getCacheRebuildWindow().isZero()
+ ? new HttpCacheRebuildCoordinator(new
ScheduledThreadPoolExecutor(httpSyncProperties.getCacheRebuildThreads(),
+ ShenyuThreadFactory.create("http-cache-rebuild",
true)),
+ httpSyncProperties.getCacheRebuildWindow().toMillis(),
httpSyncProperties.getCacheRebuildMaxPendingKeys()) : null;
+ }
+
+ @Override
+ protected void refreshCache(final ConfigGroupEnum group, final String
namespaceId, final Runnable rebuild) {
+ if (Objects.isNull(cacheRebuildCoordinator)) {
+ super.refreshCache(group, namespaceId, rebuild);
+ return;
+ }
+ Runnable work = () -> {
+ synchronized (refreshLocks.computeIfAbsent(namespaceId, key -> new
Object())) {
+ rebuild.run();
+ }
+ // The coalescing path owns its notification: wake long-poll
clients only after publication.
+ try {
+ scheduler.execute(new DataChangeTask(group, namespaceId));
+ } catch (RejectedExecutionException ex) {
+ if (!scheduler.isShutdown()) {
+ throw ex;
+ }
+ LOG.debug("skip long-polling notification for {} {}: listener
is shutting down", namespaceId, group, ex);
+ }
+ };
+ if (!cacheRebuildCoordinator.request(buildCacheKey(namespaceId,
group.name()), work)) {
+ // Bounded memory under overload: new keys apply backpressure to
their committing caller.
+ work.run();
+ }
+ }
+
+ @Override
+ protected void refreshLocalCache(final String namespaceId) {
+ synchronized (refreshLocks.computeIfAbsent(namespaceId, key -> new
Object())) {
+ super.refreshLocalCache(namespaceId);
+ }
+ }
+
+ private void notifyChanged(final ConfigGroupEnum group, final String
namespaceId) {
+ // Intentionally a no-op when coalescing is enabled: the coalesced
rebuild in refreshCache issues
+ // the DataChangeTask after it publishes the new snapshot, so
notifying here would wake clients
+ // before the cache they compare against has changed.
+ if (Objects.isNull(cacheRebuildCoordinator)) {
+ scheduler.execute(new DataChangeTask(group, namespaceId));
+ }
+ }
+
+ @Override
+ public void destroy() {
+ if (Objects.nonNull(cacheRebuildCoordinator)) {
+ cacheRebuildCoordinator.close();
+ }
+ scheduler.shutdownNow();
}
@Override
@@ -172,37 +230,37 @@ public class HttpLongPollingDataChangedListener extends
AbstractDataChangedListe
@Override
protected void afterAppAuthChanged(final List<AppAuthData> changed, final
DataEventTypeEnum eventType, final String namespaceId) {
- scheduler.execute(new DataChangeTask(ConfigGroupEnum.APP_AUTH,
namespaceId));
+ notifyChanged(ConfigGroupEnum.APP_AUTH, namespaceId);
}
@Override
protected void afterMetaDataChanged(final List<MetaData> changed, final
DataEventTypeEnum eventType, final String namespaceId) {
- scheduler.execute(new DataChangeTask(ConfigGroupEnum.META_DATA,
namespaceId));
+ notifyChanged(ConfigGroupEnum.META_DATA, namespaceId);
}
@Override
protected void afterPluginChanged(final List<PluginData> changed, final
DataEventTypeEnum eventType, final String namespaceId) {
- scheduler.execute(new DataChangeTask(ConfigGroupEnum.PLUGIN,
namespaceId));
+ notifyChanged(ConfigGroupEnum.PLUGIN, namespaceId);
}
@Override
protected void afterRuleChanged(final List<RuleData> changed, final
DataEventTypeEnum eventType, final String namespaceId) {
- scheduler.execute(new DataChangeTask(ConfigGroupEnum.RULE,
namespaceId));
+ notifyChanged(ConfigGroupEnum.RULE, namespaceId);
}
@Override
protected void afterSelectorChanged(final List<SelectorData> changed,
final DataEventTypeEnum eventType, final String namespaceId) {
- scheduler.execute(new DataChangeTask(ConfigGroupEnum.SELECTOR,
namespaceId));
+ notifyChanged(ConfigGroupEnum.SELECTOR, namespaceId);
}
@Override
protected void afterProxySelectorChanged(final List<ProxySelectorData>
changed, final DataEventTypeEnum eventType, final String namespaceId) {
- scheduler.execute(new DataChangeTask(ConfigGroupEnum.PROXY_SELECTOR,
namespaceId));
+ notifyChanged(ConfigGroupEnum.PROXY_SELECTOR, namespaceId);
}
@Override
protected void afterDiscoveryUpstreamDataChanged(final
List<DiscoverySyncData> changed, final DataEventTypeEnum eventType, final
String namespaceId) {
- scheduler.execute(new
DataChangeTask(ConfigGroupEnum.DISCOVER_UPSTREAM, namespaceId));
+ notifyChanged(ConfigGroupEnum.DISCOVER_UPSTREAM, namespaceId);
}
private List<ConfigGroupEnum> compareChangedGroup(final HttpServletRequest
request) {
@@ -426,6 +484,15 @@ public class HttpLongPollingDataChangedListener extends
AbstractDataChangedListe
@Override
public void run() {
try {
+ if (Objects.nonNull(cacheRebuildCoordinator)) {
+ // Publication may race the request thread's first
comparison. Registration and
+ // notification share this scheduler, so recheck before
entering the wait queue.
+ List<ConfigGroupEnum> changedGroups =
compareChangedGroup((HttpServletRequest) asyncContext.getRequest());
+ if (CollectionUtils.isNotEmpty(changedGroups)) {
+ sendResponse(changedGroups);
+ return;
+ }
+ }
BlockingQueue<LongPollingClient> namespaceClients =
clientsMap.getOrDefault(namespaceId, new ArrayBlockingQueue<>(1024));
this.asyncTimeoutFuture = scheduler.schedule(() -> {
namespaceClients.remove(LongPollingClient.this);
diff --git a/shenyu-admin/src/main/resources/application.yml
b/shenyu-admin/src/main/resources/application.yml
index 9265355509..015396d931 100755
--- a/shenyu-admin/src/main/resources/application.yml
+++ b/shenyu-admin/src/main/resources/application.yml
@@ -96,6 +96,9 @@ shenyu:
# connectionTimeoutMilliseconds: 2000
# http:
# enabled: true
+# cache-rebuild-window: 0ms # opt in with e.g. 10ms after measuring
convergence
+# cache-rebuild-max-pending-keys: 1024
+# cache-rebuild-threads: 1 # raise when many namespaces/groups change
concurrently
# nacos:
# url: localhost:8848
# namespace: 1c10d748-af86-43b9-8265-75f487d20c6c
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/config/properties/HttpSyncPropertiesTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/config/properties/HttpSyncPropertiesTest.java
index 4821e9781a..bf5a91d8d2 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/config/properties/HttpSyncPropertiesTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/config/properties/HttpSyncPropertiesTest.java
@@ -44,6 +44,9 @@ public final class HttpSyncPropertiesTest extends
AbstractConfigurationTest {
assertThat(httpSyncProperties.isEnabled(), comparesEqualTo(true));
Assertions.assertEquals(httpSyncProperties.getNotifyBatchSize(), 0);
assertThat(httpSyncProperties.getRefreshInterval(),
comparesEqualTo(Duration.ofMinutes(5)));
+ Assertions.assertEquals(Duration.ZERO,
httpSyncProperties.getCacheRebuildWindow());
+ Assertions.assertEquals(1024,
httpSyncProperties.getCacheRebuildMaxPendingKeys());
+ Assertions.assertEquals(1,
httpSyncProperties.getCacheRebuildThreads());
}
@Test
@@ -54,6 +57,23 @@ public final class HttpSyncPropertiesTest extends
AbstractConfigurationTest {
assertThat(httpSyncProperties.getRefreshInterval(),
comparesEqualTo(Duration.ofMinutes(1)));
}
+ @Test
+ void testCoalescingConfiguration() {
+ load(HttpSyncPropertiesConfiguration.class,
"shenyu.sync.http.cache-rebuild-window=10ms",
"shenyu.sync.http.cache-rebuild-max-pending-keys=32",
+ "shenyu.sync.http.cache-rebuild-threads=4");
+ HttpSyncProperties properties =
getContext().getBean(HttpSyncProperties.class);
+ Assertions.assertEquals(Duration.ofMillis(10),
properties.getCacheRebuildWindow());
+ Assertions.assertEquals(32,
properties.getCacheRebuildMaxPendingKeys());
+ Assertions.assertEquals(4, properties.getCacheRebuildThreads());
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
properties.setCacheRebuildThreads(0));
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
properties.setCacheRebuildThreads(65));
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
properties.setCacheRebuildWindow(Duration.ofMillis(-1)));
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
properties.setCacheRebuildWindow(Duration.ofSeconds(2)));
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
properties.setCacheRebuildWindow(Duration.ofNanos(1)));
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
properties.setCacheRebuildWindow(null));
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
properties.setCacheRebuildMaxPendingKeys(0));
+ }
+
@Configuration
@EnableConfigurationProperties(HttpSyncProperties.class)
static class HttpSyncPropertiesConfiguration {
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/http/HttpCacheRebuildBenchmark.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/http/HttpCacheRebuildBenchmark.java
new file mode 100644
index 0000000000..390170d8d7
--- /dev/null
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/http/HttpCacheRebuildBenchmark.java
@@ -0,0 +1,425 @@
+/*
+ * 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.shenyu.admin.listener.http;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import com.sun.management.ThreadMXBean;
+import com.sun.management.OperatingSystemMXBean;
+import org.apache.shenyu.admin.config.properties.ClusterProperties;
+import org.apache.shenyu.admin.config.properties.HttpSyncProperties;
+import org.apache.shenyu.admin.listener.DataChangedEvent;
+import org.apache.shenyu.admin.listener.DataChangedEventDispatcher;
+import org.apache.shenyu.admin.service.RuleService;
+import org.apache.shenyu.admin.service.SelectorService;
+import org.apache.shenyu.common.dto.RuleData;
+import org.apache.shenyu.common.dto.SelectorData;
+import org.apache.shenyu.common.enums.ConfigGroupEnum;
+import org.apache.shenyu.common.enums.DataEventTypeEnum;
+import org.h2.jdbcx.JdbcDataSource;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.ApplicationContext;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.support.TransactionSynchronization;
+import
org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import java.lang.management.ManagementFactory;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.LockSupport;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Opt-in H2 producer benchmark. Measures real queries and HTTP cache
serialization, with
+ * in-process fetch latency; it does not measure a deployed gateway's
business-request p99.
+ */
+@EnabledIfSystemProperty(named = "shenyu.http.benchmark", matches = "true")
+class HttpCacheRebuildBenchmark {
+
+ private static final ThreadMXBean THREADS = (ThreadMXBean)
ManagementFactory.getThreadMXBean();
+
+ private static final OperatingSystemMXBean PROCESS =
(OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean();
+
+ private final int rows = Integer.getInteger("shenyu.http.benchmark.rows",
1000);
+
+ private final int events =
Integer.getInteger("shenyu.http.benchmark.events", 100);
+
+ private final int warmup =
Integer.getInteger("shenyu.http.benchmark.warmup", 5);
+
+ private final int samples =
Integer.getInteger("shenyu.http.benchmark.samples", 3);
+
+ private final int handleChars =
Integer.getInteger("shenyu.http.benchmark.handleChars", 256);
+
+ @Test
+ void benchmark() throws Exception {
+ THREADS.setThreadAllocatedMemoryEnabled(true);
+ Logger root = (Logger)
LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME);
+ Level previous = root.getLevel();
+ root.setLevel(Level.WARN);
+ List<String> output = new ArrayList<>();
+
output.add("namespaces,groups,pace_ms,window_ms,sample,rows_per_key,handle_chars,events,queries,serializations,hashes,"
+ +
"allocated_bytes,cpu_ms,gc_count,gc_ms,elapsed_ms,edit_p99_ms,fetch_p99_ms,max_publish_lag_ms,final_convergence_ms");
+ try {
+ for (int namespaces : List.of(1, 2)) {
+ for (int groups : List.of(1, 2)) {
+ for (int pace : List.of(0, 5, 25)) {
+ compare(namespaces, groups, pace, output);
+ }
+ }
+ }
+ Path path =
Path.of(System.getProperty("shenyu.http.benchmark.output",
"target/http-cache-rebuild-benchmark.csv"));
+ Files.createDirectories(path.toAbsolutePath().getParent());
+ Files.write(path, output);
+ } finally {
+ root.setLevel(previous);
+ }
+ LoggerFactory.getLogger(getClass()).info("HTTP cache benchmark: JDK
{}, processors {}, max heap {}, JVM args {}",
+ System.getProperty("java.runtime.version"),
Runtime.getRuntime().availableProcessors(),
+ Runtime.getRuntime().maxMemory(),
ManagementFactory.getRuntimeMXBean().getInputArguments());
+ }
+
+ private void compare(final int namespaces, final int groups, final int
pace, final List<String> output) throws Exception {
+ // Warm both variants independently; alternate measurement order to
expose ordering effects.
+ for (int sample = 0; sample < samples; sample++) {
+ for (int window : sample % 2 == 0 ? List.of(0, 10) : List.of(10,
0)) {
+ try (Fixture fixture = new Fixture(namespaces, groups,
window)) {
+ for (int iteration = 0; iteration < warmup; iteration++) {
+ fixture.run(events, 0);
+ }
+ for (int iteration = 0; iteration < 20; iteration++) {
+ fixture.keys.forEach(key ->
fixture.listener.fetchConfig(key.group, key.namespace));
+ }
+ output.add(fixture.measure(pace, window, sample));
+ }
+ }
+ }
+ }
+
+ private static long gcCount() {
+ return
ManagementFactory.getGarbageCollectorMXBeans().stream().mapToLong(bean ->
Math.max(0, bean.getCollectionCount())).sum();
+ }
+
+ private static long gcMillis() {
+ return
ManagementFactory.getGarbageCollectorMXBeans().stream().mapToLong(bean ->
Math.max(0, bean.getCollectionTime())).sum();
+ }
+
+ private static double p99(final List<Long> samples) {
+ List<Long> sorted = new ArrayList<>(samples);
+ Collections.sort(sorted);
+ return sorted.isEmpty() ? 0 : sorted.get(Math.min(sorted.size() - 1,
(int) Math.ceil(sorted.size() * 0.99) - 1)) / 1_000_000.0;
+ }
+
+ private final class Fixture implements AutoCloseable {
+
+ private final JdbcTemplate jdbc;
+
+ private final TransactionTemplate transactions;
+
+ private final MeasuredListener listener;
+
+ private final DataChangedEventDispatcher dispatcher;
+
+ private final List<Key> keys = new ArrayList<>();
+
+ private final AtomicInteger queries = new AtomicInteger();
+
+ private final Map<String, String> expected = new ConcurrentHashMap<>();
+
+ private final Map<String, Long> committed = new ConcurrentHashMap<>();
+
+ private int sequence;
+
+ private Fixture(final int namespaces, final int groups, final int
window) {
+ JdbcDataSource source = new JdbcDataSource();
+ source.setURL("jdbc:h2:mem:" + UUID.randomUUID() +
";DB_CLOSE_DELAY=-1");
+ jdbc = new JdbcTemplate(source);
+ transactions = new TransactionTemplate(new
DataSourceTransactionManager(source));
+ jdbc.execute("CREATE TABLE config (namespace_id VARCHAR,
group_name VARCHAR, id INT, name VARCHAR, handle VARCHAR, PRIMARY
KEY(namespace_id, group_name, id))");
+ for (int namespace = 0; namespace < namespaces; namespace++) {
+ for (ConfigGroupEnum group : List.of(ConfigGroupEnum.RULE,
ConfigGroupEnum.SELECTOR).subList(0, groups)) {
+ Key key = new Key("namespace-" + namespace, group);
+ keys.add(key);
+ List<Object[]> batch = new ArrayList<>();
+ for (int row = 0; row < rows; row++) {
+ batch.add(new Object[]{key.namespace, group.name(),
row, "initial", "x".repeat(handleChars)});
+ }
+ jdbc.batchUpdate("INSERT INTO config VALUES (?, ?, ?, ?,
?)", batch);
+ }
+ }
+ HttpSyncProperties properties = new HttpSyncProperties();
+ properties.setCacheRebuildWindow(Duration.ofMillis(window));
+ listener = new MeasuredListener(properties, committed);
+ listener.clearCache();
+ RuleService rules = mock(RuleService.class);
+
when(rules.listAllByNamespaceId(anyString())).thenAnswer(invocation ->
loadRules(invocation.getArgument(0)));
+ SelectorService selectors = mock(SelectorService.class);
+
when(selectors.listAllByNamespaceId(anyString())).thenAnswer(invocation ->
loadSelectors(invocation.getArgument(0)));
+ ReflectionTestUtils.setField(listener, "ruleService", rules);
+ ReflectionTestUtils.setField(listener, "selectorService",
selectors);
+ dispatcher = new
DataChangedEventDispatcher(mock(ApplicationContext.class));
+ ReflectionTestUtils.setField(dispatcher, "listeners",
List.of(listener));
+ ReflectionTestUtils.setField(dispatcher, "clusterProperties", new
ClusterProperties());
+ }
+
+ private List<RuleData> loadRules(final String namespace) {
+ queries.incrementAndGet();
+ return jdbc.query("SELECT * FROM config WHERE namespace_id = ? AND
group_name = 'RULE' ORDER BY id", (row, index) -> {
+ RuleData data = new RuleData();
+ data.setId(row.getString("id"));
+ data.setName(row.getString("name"));
+ data.setHandle(row.getString("handle"));
+ data.setNamespaceId(namespace);
+ return data;
+ }, namespace);
+ }
+
+ private List<SelectorData> loadSelectors(final String namespace) {
+ queries.incrementAndGet();
+ return jdbc.query("SELECT * FROM config WHERE namespace_id = ? AND
group_name = 'SELECTOR' ORDER BY id", (row, index) -> {
+ SelectorData data = new SelectorData();
+ data.setId(row.getString("id"));
+ data.setName(row.getString("name"));
+ data.setHandle(row.getString("handle"));
+ data.setNamespaceId(namespace);
+ return data;
+ }, namespace);
+ }
+
+ private void write(final Key key) {
+ String version = "version-" + ++sequence;
+ transactions.executeWithoutResult(status -> {
+ jdbc.update("UPDATE config SET name = ? WHERE namespace_id = ?
AND group_name = ? AND id = 0", version, key.namespace, key.group.name());
+ TransactionSynchronizationManager.registerSynchronization(new
TransactionSynchronization() {
+ @Override
+ public void afterCommit() {
+ committed.put(version, System.nanoTime());
+ expected.put(key.cacheKey(), version);
+ }
+ });
+ if (key.group == ConfigGroupEnum.RULE) {
+ RuleData data = new RuleData();
+ data.setNamespaceId(key.namespace);
+ dispatcher.onApplicationEvent(new
DataChangedEvent(key.group, DataEventTypeEnum.UPDATE, List.of(data)));
+ } else {
+ SelectorData data = new SelectorData();
+ data.setNamespaceId(key.namespace);
+ dispatcher.onApplicationEvent(new
DataChangedEvent(key.group, DataEventTypeEnum.UPDATE, List.of(data)));
+ }
+ });
+ }
+
+ private RunResult run(final int count, final int pace) throws
Exception {
+ List<Long> edits = new ArrayList<>();
+ long next = System.nanoTime();
+ for (int index = 0; index < count; index++) {
+ LockSupport.parkNanos(Math.max(0, next - System.nanoTime()));
+ long start = System.nanoTime();
+ write(keys.get(index % keys.size()));
+ edits.add(System.nanoTime() - start);
+ next += TimeUnit.MILLISECONDS.toNanos(pace);
+ }
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (!expected.entrySet().stream().allMatch(entry ->
entry.getValue().equals(listener.published.get(entry.getKey())))) {
+ assertTrue(System.nanoTime() < deadline, "cache did not
converge");
+ Thread.sleep(1);
+ }
+ long lag = expected.values().stream().mapToLong(version ->
listener.publishTimes.get(version) - committed.get(version)).max().orElse(0);
+ return new RunResult(edits, lag);
+ }
+
+ private String measure(final int pace, final int window, final int
sample) throws Exception {
+ // Drain any redundant generation from warmup before counters are
reset.
+ ScheduledBarrier.await(listener);
+ queries.set(0);
+ listener.serializations.set(0);
+ listener.maxLag.set(0);
+ AtomicBoolean reading = new AtomicBoolean(true);
+ List<Long> fetches = new ArrayList<>();
+ CountDownLatch readerDone = new CountDownLatch(1);
+ Thread reader = new Thread(() -> {
+ while (reading.get()) {
+ for (Key key : keys) {
+ long start = System.nanoTime();
+ listener.fetchConfig(key.group, key.namespace);
+ fetches.add(System.nanoTime() - start);
+ }
+ LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(2));
+ }
+ try {
+ readerDone.await();
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ }
+ }, "http-benchmark-reader");
+ reader.start();
+ Map<Long, Long> before = threadAllocations();
+ final long cpuBefore = PROCESS.getProcessCpuTime();
+ final long gc = gcCount();
+ final long gcTime = gcMillis();
+ long start = System.nanoTime();
+ RunResult result;
+ try {
+ result = run(events, pace);
+ ScheduledBarrier.await(listener);
+ } catch (Exception | AssertionError ex) {
+ readerDone.countDown();
+ throw ex;
+ } finally {
+ reading.set(false);
+ }
+ final long elapsed = System.nanoTime() - start;
+ // The reader stays alive until sampled, to include its
allocations and CPU.
+ final long cpu = PROCESS.getProcessCpuTime() - cpuBefore;
+ Map<Long, Long> after = threadAllocations();
+ readerDone.countDown();
+ reader.join(5000);
+ long allocation = 0;
+ for (Map.Entry<Long, Long> entry : after.entrySet()) {
+ allocation += Math.max(0, entry.getValue() -
before.getOrDefault(entry.getKey(), 0L));
+ }
+ assertEquals(queries.get(), listener.serializations.get());
+ assertTrue(result.finalLag <= TimeUnit.MILLISECONDS.toNanos(250),
"250ms final publication budget exceeded");
+ return String.format(Locale.ROOT,
"%d,%d,%d,%d,%d,%d,%d,%d,%d,%d,%d,%d,%.3f,%d,%d,%.3f,%.3f,%.3f,%.3f,%.3f",
+ keys.stream().map(key ->
key.namespace).distinct().count(), keys.stream().map(key ->
key.group).distinct().count(),
+ pace, window, sample, rows, handleChars, events,
queries.get(), listener.serializations.get(), listener.serializations.get(),
+ allocation, cpu / 1_000_000.0, gcCount() - gc, gcMillis()
- gcTime, elapsed / 1_000_000.0,
+ p99(result.edits), p99(fetches), listener.maxLag.get() /
1_000_000.0, result.finalLag / 1_000_000.0);
+ }
+
+ private Map<Long, Long> threadAllocations() {
+ Map<Long, Long> result = new HashMap<>();
+ for (long id : THREADS.getAllThreadIds()) {
+ result.put(id, Math.max(0,
THREADS.getThreadAllocatedBytes(id)));
+ }
+ return result;
+ }
+
+ @Override
+ public void close() {
+ listener.destroy();
+ listener.clearCache();
+ jdbc.execute("SHUTDOWN");
+ }
+ }
+
+ private static final class ScheduledBarrier {
+
+ private static void await(final MeasuredListener listener) throws
Exception {
+ HttpCacheRebuildCoordinator coordinator =
(HttpCacheRebuildCoordinator) ReflectionTestUtils.getField(listener,
"cacheRebuildCoordinator");
+ if (java.util.Objects.nonNull(coordinator)) {
+ Map<?, ?> pending = (Map<?, ?>)
ReflectionTestUtils.getField(coordinator, "pending");
+ long deadline = System.nanoTime() +
TimeUnit.SECONDS.toNanos(10);
+ while (true) {
+ synchronized (pending) {
+ if (pending.isEmpty()) {
+ return;
+ }
+ }
+ assertTrue(System.nanoTime() < deadline, "pending rebuilds
did not drain");
+ Thread.sleep(1);
+ }
+ }
+ }
+ }
+
+ private static final class MeasuredListener extends
HttpLongPollingDataChangedListener {
+
+ private final Map<String, Long> committed;
+
+ private final Map<String, String> published = new
ConcurrentHashMap<>();
+
+ private final Map<String, Long> publishTimes = new
ConcurrentHashMap<>();
+
+ private final AtomicInteger serializations = new AtomicInteger();
+
+ private final AtomicLong maxLag = new AtomicLong();
+
+ private MeasuredListener(final HttpSyncProperties properties, final
Map<String, Long> committed) {
+ super(properties);
+ this.committed = committed;
+ }
+
+ @Override
+ protected <T> void updateCache(final ConfigGroupEnum group, final
List<T> data, final String namespaceId) {
+ super.updateCache(group, data, namespaceId);
+ serializations.incrementAndGet();
+ String version = group == ConfigGroupEnum.RULE ? ((RuleData)
data.get(0)).getName() : ((SelectorData) data.get(0)).getName();
+ long now = System.nanoTime();
+ if (java.util.Objects.isNull(publishTimes.putIfAbsent(version,
now))) {
+ maxLag.accumulateAndGet(now - committed.getOrDefault(version,
now), Math::max);
+ }
+ published.put(buildCacheKey(namespaceId, group.name()), version);
+ }
+
+ private void clearCache() {
+ CACHE.clear();
+ }
+ }
+
+ private static final class Key {
+
+ private final String namespace;
+
+ private final ConfigGroupEnum group;
+
+ private Key(final String namespace, final ConfigGroupEnum group) {
+ this.namespace = namespace;
+ this.group = group;
+ }
+
+ private String cacheKey() {
+ return HttpLongPollingDataChangedListener.buildCacheKey(namespace,
group.name());
+ }
+ }
+
+ private static final class RunResult {
+
+ private final List<Long> edits;
+
+ private final long finalLag;
+
+ private RunResult(final List<Long> edits, final long finalLag) {
+ this.edits = edits;
+ this.finalLag = finalLag;
+ }
+ }
+}
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/http/HttpCacheRebuildCoordinatorTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/http/HttpCacheRebuildCoordinatorTest.java
new file mode 100644
index 0000000000..6680e5c196
--- /dev/null
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/http/HttpCacheRebuildCoordinatorTest.java
@@ -0,0 +1,296 @@
+/*
+ * 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.shenyu.admin.listener.http;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Deterministic tests of deadlines, generations, retries and bounded
admission.
+ */
+class HttpCacheRebuildCoordinatorTest {
+
+ private ScheduledExecutorService executor;
+
+ private Queue<Runnable> tasks;
+
+ private List<Long> delays;
+
+ private HttpCacheRebuildCoordinator coordinator;
+
+ @BeforeEach
+ void setUp() {
+ executor = mock(ScheduledExecutorService.class);
+ tasks = new ArrayDeque<>();
+ delays = new ArrayList<>();
+ doAnswer(invocation -> {
+ tasks.add(invocation.getArgument(0));
+ delays.add(invocation.getArgument(1));
+ return null;
+ }).when(executor).schedule(any(Runnable.class), anyLong(),
eq(TimeUnit.MILLISECONDS));
+ coordinator = new HttpCacheRebuildCoordinator(executor, 10, 2);
+ }
+
+ @Test
+ void burstKeepsFirstDeadlineAndReleasesIdleKey() {
+ AtomicInteger rebuilds = new AtomicInteger();
+ for (int index = 0; index < 1000; index++) {
+ assertTrue(coordinator.request("a_RULE",
rebuilds::incrementAndGet));
+ }
+ verify(executor, times(1)).schedule(any(Runnable.class), eq(10L),
eq(TimeUnit.MILLISECONDS));
+ tasks.remove().run();
+ assertEquals(1, rebuilds.get());
+ assertTrue(tasks.isEmpty());
+ assertTrue(coordinator.request("a_RULE", rebuilds::incrementAndGet));
+ tasks.remove().run();
+ assertEquals(2, rebuilds.get());
+ }
+
+ @Test
+ void concurrentArrivalDuringRebuildIsNotLost() throws Exception {
+ CountDownLatch reading = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ AtomicInteger database = new AtomicInteger(1);
+ AtomicInteger published = new AtomicInteger();
+ AtomicInteger calls = new AtomicInteger();
+ Runnable rebuild = () -> {
+ int snapshot = database.get();
+ if (calls.incrementAndGet() == 1) {
+ reading.countDown();
+ try {
+ assertTrue(release.await(5, TimeUnit.SECONDS));
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(ex);
+ }
+ }
+ published.set(snapshot);
+ };
+ coordinator.request("a_RULE", rebuild);
+ FutureTask<Void> first = new FutureTask<>(tasks.remove(), null);
+ Thread worker = new Thread(first);
+ worker.start();
+ try {
+ assertTrue(reading.await(5, TimeUnit.SECONDS));
+ database.set(2);
+ coordinator.request("a_RULE", rebuild);
+ } finally {
+ release.countDown();
+ }
+ first.get(5, TimeUnit.SECONDS);
+ assertEquals(1, published.get());
+ assertEquals(1, tasks.size());
+ tasks.remove().run();
+ assertEquals(2, published.get());
+ assertEquals(2, calls.get());
+ assertTrue(tasks.isEmpty());
+ }
+
+ @Test
+ void separatesKeysAndBoundsPendingWork() {
+ AtomicInteger rebuilds = new AtomicInteger();
+ assertTrue(coordinator.request("a_RULE", rebuilds::incrementAndGet));
+ assertTrue(coordinator.request("b_RULE", rebuilds::incrementAndGet));
+ assertFalse(coordinator.request("a_SELECTOR",
rebuilds::incrementAndGet));
+ assertTrue(coordinator.request("a_RULE", rebuilds::incrementAndGet));
+ assertEquals(2, tasks.size());
+ tasks.remove().run();
+ assertTrue(coordinator.request("a_SELECTOR",
rebuilds::incrementAndGet));
+ tasks.remove().run();
+ tasks.remove().run();
+ assertEquals(3, rebuilds.get());
+ assertTrue(tasks.isEmpty());
+ }
+
+ @Test
+ void failureRetriesWithoutAnotherEventAndYieldsToOtherKeys() {
+ AtomicInteger calls = new AtomicInteger();
+ AtomicInteger other = new AtomicInteger();
+ coordinator.request("a_RULE", () -> {
+ if (calls.incrementAndGet() == 1) {
+ throw new IllegalStateException("temporary database failure");
+ }
+ });
+ coordinator.request("b_RULE", other::incrementAndGet);
+ tasks.remove().run();
+ assertEquals(2, tasks.size());
+ verify(executor).schedule(any(Runnable.class), eq(1000L),
eq(TimeUnit.MILLISECONDS));
+ tasks.remove().run();
+ assertEquals(1, other.get());
+ tasks.remove().run();
+ assertEquals(2, calls.get());
+ assertTrue(tasks.isEmpty());
+ }
+
+ @Test
+ void rejectedSubmissionDoesNotLeavePendingMarker() {
+ when(executor.schedule(any(Runnable.class), eq(10L),
eq(TimeUnit.MILLISECONDS)))
+ .thenThrow(new RejectedExecutionException());
+ assertFalse(coordinator.request("a_RULE", () -> { }));
+ assertFalse(coordinator.request("a_RULE", () -> { }));
+ verify(executor, times(2)).schedule(any(Runnable.class), eq(10L),
eq(TimeUnit.MILLISECONDS));
+ }
+
+ @Test
+ void closeCancelsPendingRebuilds() {
+ AtomicInteger rebuilds = new AtomicInteger();
+ coordinator.request("a_RULE", rebuilds::incrementAndGet);
+ coordinator.close();
+ tasks.remove().run();
+ assertEquals(0, rebuilds.get());
+ verify(executor).shutdownNow();
+ // A commit racing shutdown is told to rebuild synchronously instead
of failing its hook.
+ assertFalse(coordinator.request("a_RULE", rebuilds::incrementAndGet));
+ assertTrue(tasks.isEmpty());
+ }
+
+ @Test
+ void
consecutiveFailuresBackOffExponentiallyUpToTheCapAndResetAfterSuccess() {
+ AtomicInteger calls = new AtomicInteger();
+ AtomicInteger failuresLeft = new AtomicInteger(7);
+ Runnable rebuild = () -> {
+ calls.incrementAndGet();
+ if (failuresLeft.getAndDecrement() > 0) {
+ throw new IllegalStateException("database unavailable");
+ }
+ };
+ coordinator.request("a_RULE", rebuild);
+ while (!tasks.isEmpty()) {
+ tasks.remove().run();
+ }
+ assertEquals(8, calls.get());
+ assertEquals(List.of(10L, 1000L, 2000L, 4000L, 8000L, 16000L, 30000L,
30000L), delays);
+
+ delays.clear();
+ failuresLeft.set(1);
+ coordinator.request("a_RULE", rebuild);
+ while (!tasks.isEmpty()) {
+ tasks.remove().run();
+ }
+ // The previous success reset the failure count, so backoff starts
again at one second.
+ assertEquals(List.of(10L, 1000L), delays);
+ }
+
+ @Test
+ void retryDelayNeverOverflowsAndNeverUndercutsTheWindow() {
+ assertEquals(HttpCacheRebuildCoordinator.FAILURE_RETRY_MILLIS,
coordinator.retryDelayMillis(1));
+ assertEquals(HttpCacheRebuildCoordinator.MAX_FAILURE_RETRY_MILLIS,
coordinator.retryDelayMillis(Integer.MAX_VALUE));
+ HttpCacheRebuildCoordinator slowWindow = new
HttpCacheRebuildCoordinator(executor, 5000, 2);
+ assertEquals(5000, slowWindow.retryDelayMillis(1));
+ assertEquals(HttpCacheRebuildCoordinator.MAX_FAILURE_RETRY_MILLIS,
slowWindow.retryDelayMillis(10));
+ }
+
+ @Test
+ void multipleWorkersRebuildDistinctKeysInParallel() throws Exception {
+ ScheduledThreadPoolExecutor workers = new
ScheduledThreadPoolExecutor(2);
+ HttpCacheRebuildCoordinator parallel = new
HttpCacheRebuildCoordinator(workers, 1, 16);
+ try {
+ // Each rebuild waits for the other key: this only completes when
both run at the same time.
+ CyclicBarrier bothRunning = new CyclicBarrier(2);
+ CountDownLatch done = new CountDownLatch(2);
+ Runnable rebuild = () -> {
+ try {
+ bothRunning.await(5, TimeUnit.SECONDS);
+ done.countDown();
+ } catch (Exception ex) {
+ throw new IllegalStateException(ex);
+ }
+ };
+ assertTrue(parallel.request("a_RULE", rebuild));
+ assertTrue(parallel.request("b_RULE", rebuild));
+ assertTrue(done.await(5, TimeUnit.SECONDS));
+ } finally {
+ parallel.close();
+ }
+ }
+
+ @Test
+ void multipleWorkersNeverRebuildTheSameKeyConcurrently() throws Exception {
+ ScheduledThreadPoolExecutor workers = new
ScheduledThreadPoolExecutor(4);
+ HttpCacheRebuildCoordinator parallel = new
HttpCacheRebuildCoordinator(workers, 1, 16);
+ AtomicInteger running = new AtomicInteger();
+ AtomicInteger maxRunning = new AtomicInteger();
+ AtomicInteger database = new AtomicInteger();
+ AtomicInteger published = new AtomicInteger(-1);
+ Runnable rebuild = () -> {
+ maxRunning.accumulateAndGet(running.incrementAndGet(), Math::max);
+ try {
+ int snapshot = database.get();
+ Thread.sleep(1);
+ published.set(snapshot);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ } finally {
+ running.decrementAndGet();
+ }
+ };
+ ExecutorService writers = Executors.newFixedThreadPool(4);
+ try {
+ List<Future<?>> commits = new ArrayList<>();
+ for (int index = 1; index <= 200; index++) {
+ int value = index;
+ commits.add(writers.submit(() -> {
+ database.accumulateAndGet(value, Math::max);
+ if (!parallel.request("a_RULE", rebuild)) {
+ rebuild.run();
+ }
+ }));
+ }
+ for (Future<?> commit : commits) {
+ commit.get(5, TimeUnit.SECONDS);
+ }
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ while (published.get() != 200 && System.nanoTime() < deadline) {
+ Thread.sleep(5);
+ }
+ assertEquals(200, published.get());
+ assertEquals(1, maxRunning.get());
+ } finally {
+ writers.shutdownNow();
+ parallel.close();
+ }
+ }
+}
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/http/HttpCacheRebuildTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/http/HttpCacheRebuildTest.java
new file mode 100644
index 0000000000..9bfab8b654
--- /dev/null
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/http/HttpCacheRebuildTest.java
@@ -0,0 +1,357 @@
+/*
+ * 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.shenyu.admin.listener.http;
+
+import jakarta.servlet.AsyncContext;
+import org.apache.shenyu.admin.config.properties.ClusterProperties;
+import org.apache.shenyu.admin.config.properties.HttpSyncProperties;
+import org.apache.shenyu.admin.listener.ConfigDataCache;
+import org.apache.shenyu.admin.listener.DataChangedEvent;
+import org.apache.shenyu.admin.listener.DataChangedEventDispatcher;
+import org.apache.shenyu.admin.listener.DataChangedListener;
+import org.apache.shenyu.admin.service.RuleService;
+import org.apache.shenyu.admin.service.SelectorService;
+import org.apache.shenyu.common.dto.RuleData;
+import org.apache.shenyu.common.dto.SelectorData;
+import org.apache.shenyu.common.enums.ConfigGroupEnum;
+import org.apache.shenyu.common.enums.DataEventTypeEnum;
+import org.h2.jdbcx.JdbcDataSource;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.context.ApplicationContext;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.mock.web.MockHttpServletRequest;
+import org.springframework.mock.web.MockHttpServletResponse;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Queue;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Exercise committed independent events against H2, the real dispatcher and
the HTTP cache.
+ */
+class HttpCacheRebuildTest {
+
+ private TestListener listener;
+
+ private JdbcTemplate jdbc;
+
+ private TransactionTemplate transactions;
+
+ private DataChangedEventDispatcher dispatcher;
+
+ private DataChangedListener otherListener;
+
+ private Queue<Runnable> tasks;
+
+ private final AtomicInteger queries = new AtomicInteger();
+
+ @BeforeEach
+ void setUp() {
+ JdbcDataSource source = new JdbcDataSource();
+ source.setURL("jdbc:h2:mem:" + UUID.randomUUID() +
";DB_CLOSE_DELAY=-1");
+ jdbc = new JdbcTemplate(source);
+ jdbc.execute("CREATE TABLE test_rule (id VARCHAR PRIMARY KEY,
namespace_id VARCHAR, name VARCHAR)");
+ transactions = new TransactionTemplate(new
DataSourceTransactionManager(source));
+ HttpSyncProperties properties = new HttpSyncProperties();
+ properties.setCacheRebuildWindow(Duration.ofMillis(10));
+ listener = new TestListener(properties);
+ listener.clearCache();
+ HttpCacheRebuildCoordinator original = (HttpCacheRebuildCoordinator)
ReflectionTestUtils.getField(listener, "cacheRebuildCoordinator");
+ original.close();
+ ScheduledExecutorService executor =
mock(ScheduledExecutorService.class);
+ tasks = new ConcurrentLinkedQueue<>();
+ doAnswer(invocation -> {
+ tasks.add(invocation.getArgument(0));
+ return null;
+ }).when(executor).schedule(any(Runnable.class), eq(10L),
eq(TimeUnit.MILLISECONDS));
+ ReflectionTestUtils.setField(listener, "cacheRebuildCoordinator", new
HttpCacheRebuildCoordinator(executor, 10, 2));
+ RuleService rules = mock(RuleService.class);
+ when(rules.listAllByNamespaceId(anyString())).thenAnswer(invocation ->
{
+ queries.incrementAndGet();
+ String namespaceId = invocation.getArgument(0);
+ return jdbc.query("SELECT id, namespace_id, name FROM test_rule
WHERE namespace_id = ? ORDER BY id",
+ (row, index) -> rule(row.getString("id"),
row.getString("namespace_id"), row.getString("name")), namespaceId);
+ });
+ ReflectionTestUtils.setField(listener, "ruleService", rules);
+ SelectorService selectors = mock(SelectorService.class);
+
when(selectors.listAllByNamespaceId(anyString())).thenReturn(Collections.emptyList());
+ ReflectionTestUtils.setField(listener, "selectorService", selectors);
+ otherListener = mock(DataChangedListener.class);
+ dispatcher = new
DataChangedEventDispatcher(mock(ApplicationContext.class));
+ ReflectionTestUtils.setField(dispatcher, "clusterProperties", new
ClusterProperties());
+ ReflectionTestUtils.setField(dispatcher, "listeners",
List.of(listener, otherListener));
+ }
+
+ @AfterEach
+ void tearDown() {
+ listener.destroy();
+ listener.clearCache();
+ jdbc.execute("SHUTDOWN");
+ }
+
+ @Test
+ void independentConcurrentCommitsCoalesceWithoutDroppingEvents() throws
Exception {
+ ExecutorService writers = Executors.newFixedThreadPool(4);
+ try {
+ List<Future<?>> commits = new ArrayList<>();
+ for (int index = 0; index < 40; index++) {
+ String id = String.valueOf(index);
+ commits.add(writers.submit(() -> commit(id, "a",
"committed")));
+ }
+ for (Future<?> commit : commits) {
+ commit.get(10, TimeUnit.SECONDS);
+ }
+ } finally {
+ writers.shutdownNow();
+ }
+ assertEquals(0, queries.get());
+ assertEquals(1, tasks.size());
+ assertEquals(40, listener.callbacks.size());
+ tasks.remove().run();
+ assertEquals(1, queries.get());
+ assertEquals(1, listener.serializations.get());
+ assertEquals(40, listener.fetchConfig(ConfigGroupEnum.RULE,
"a").getData().size());
+ verify(otherListener,
org.mockito.Mockito.times(40)).onRuleChanged(any(),
eq(DataEventTypeEnum.UPDATE), any());
+ }
+
+ @Test
+ void rollbackNeverSchedulesOrPublishesUncommittedData() {
+ transactions.executeWithoutResult(status -> {
+ jdbc.update("INSERT INTO test_rule VALUES (?, ?, ?)", "1", "a",
"rolled-back");
+ event(rule("1", "a", "rolled-back"), DataEventTypeEnum.UPDATE);
+ assertTrue(tasks.isEmpty());
+ assertEquals(0, queries.get());
+ status.setRollbackOnly();
+ });
+ assertTrue(tasks.isEmpty());
+ assertTrue(listener.callbacks.isEmpty());
+ verify(otherListener, never()).onRuleChanged(any(), any(), any());
+ commit("2", "a", "committed");
+ tasks.remove().run();
+ assertEquals("committed", ((RuleData)
listener.fetchConfig(ConfigGroupEnum.RULE, "a").getData().get(0)).getName());
+ }
+
+ @Test
+ void deleteThenUpdateUsesAuthoritativeSnapshotAndEmptyGroup() {
+ commit("1", "a", "first");
+ tasks.remove().run();
+ transactions.executeWithoutResult(status -> {
+ jdbc.update("DELETE FROM test_rule WHERE id = '1'");
+ event(rule("1", "a", "first"), DataEventTypeEnum.DELETE);
+ });
+ commit("1", "a", "replacement");
+ tasks.remove().run();
+ assertEquals("replacement", ((RuleData)
listener.fetchConfig(ConfigGroupEnum.RULE, "a").getData().get(0)).getName());
+ transactions.executeWithoutResult(status -> {
+ jdbc.update("DELETE FROM test_rule");
+ event(rule("1", "a", "replacement"), DataEventTypeEnum.DELETE);
+ });
+ tasks.remove().run();
+ assertTrue(listener.fetchConfig(ConfigGroupEnum.RULE,
"a").getData().isEmpty());
+ assertEquals(List.of(DataEventTypeEnum.UPDATE,
DataEventTypeEnum.DELETE, DataEventTypeEnum.UPDATE, DataEventTypeEnum.DELETE),
listener.callbacks);
+ }
+
+ @Test
+ void namespacesGroupsAndCapacityFallbackRemainIndependent() {
+ commit("1", "a", "a");
+ commit("2", "b", "b");
+ SelectorData selector = new SelectorData();
+ selector.setNamespaceId("a");
+ listener.onSelectorChanged(List.of(selector),
DataEventTypeEnum.UPDATE);
+ // Both pending slots are occupied. The new selector key rebuilds on
the caller.
+ assertTrue(listener.fetchConfig(ConfigGroupEnum.SELECTOR,
"a").getData().isEmpty());
+ assertEquals(2, tasks.size());
+ tasks.remove().run();
+ tasks.remove().run();
+ assertEquals("a", ((RuleData)
listener.fetchConfig(ConfigGroupEnum.RULE, "a").getData().get(0)).getName());
+ assertEquals("b", ((RuleData)
listener.fetchConfig(ConfigGroupEnum.RULE, "b").getData().get(0)).getName());
+ }
+
+ @Test
+ void unchangedDigestPreservesVersion() {
+ commit("1", "a", "same");
+ tasks.remove().run();
+ ConfigDataCache before = listener.cached("a");
+ event(rule("1", "a", "same"), DataEventTypeEnum.UPDATE);
+ tasks.remove().run();
+ assertSame(before, listener.cached("a"));
+ }
+
+ @Test
+ void disabledModeStillRebuildsSynchronouslyPerEvent() {
+ ((HttpCacheRebuildCoordinator) ReflectionTestUtils.getField(listener,
"cacheRebuildCoordinator")).close();
+ ReflectionTestUtils.setField(listener, "cacheRebuildCoordinator",
null);
+ commit("1", "a", "first");
+ commit("2", "a", "second");
+ assertEquals(2, queries.get());
+ assertEquals(2, listener.serializations.get());
+ assertEquals(2, listener.fetchConfig(ConfigGroupEnum.RULE,
"a").getData().size());
+ assertTrue(tasks.isEmpty());
+ }
+
+ @Test
+ void commitRacingShutdownRebuildsSynchronouslyWithoutFailingTheCommit() {
+ listener.destroy();
+ commit("1", "a", "after-shutdown");
+ assertTrue(tasks.isEmpty());
+ assertEquals(1, queries.get());
+ assertEquals("after-shutdown", ((RuleData)
listener.fetchConfig(ConfigGroupEnum.RULE, "a").getData().get(0)).getName());
+ assertEquals(List.of(DataEventTypeEnum.UPDATE), listener.callbacks);
+ }
+
+ @Test
+ void configuredWorkerThreadsSizeTheRebuildExecutor() {
+ HttpSyncProperties properties = new HttpSyncProperties();
+ properties.setCacheRebuildWindow(Duration.ofMillis(10));
+ properties.setCacheRebuildThreads(3);
+ TestListener sized = new TestListener(properties);
+ try {
+ HttpCacheRebuildCoordinator coordinator =
(HttpCacheRebuildCoordinator) ReflectionTestUtils.getField(sized,
"cacheRebuildCoordinator");
+ assertEquals(3,
((java.util.concurrent.ScheduledThreadPoolExecutor)
ReflectionTestUtils.getField(coordinator, "executor")).getCorePoolSize());
+ } finally {
+ sized.destroy();
+ }
+ }
+
+ @Test
+ void longPollIsWokenOnlyAfterPublicationAndOnlyForItsNamespace() throws
Exception {
+ AsyncContext client = client("a");
+ final AsyncContext other = client("b");
+ scheduler().submit(() -> { }).get(5, TimeUnit.SECONDS);
+ commit("1", "a", "visible-before-notification");
+ scheduler().submit(() -> { }).get(5, TimeUnit.SECONDS);
+ verify(client, never()).complete();
+ tasks.remove().run();
+ scheduler().submit(() -> { }).get(5, TimeUnit.SECONDS);
+ verify(client).complete();
+ verify(other, never()).complete();
+ assertTrue(((MockHttpServletResponse)
client.getResponse()).getContentAsString().contains("RULE"));
+ assertEquals(1, listener.fetchConfig(ConfigGroupEnum.RULE,
"a").getData().size());
+ }
+
+ private AsyncContext client(final String namespace) {
+ AsyncContext context = mock(AsyncContext.class);
+ MockHttpServletRequest request = new MockHttpServletRequest();
+ request.setParameter("namespaceId", namespace);
+ for (ConfigGroupEnum group : ConfigGroupEnum.values()) {
+ listener.seedCache(group, namespace);
+ request.setParameter(group.name(), "seed,0");
+ }
+ when(context.getRequest()).thenReturn(request);
+ when(context.getResponse()).thenReturn(new MockHttpServletResponse());
+ scheduler().execute(listener.new LongPollingClient(context,
"127.0.0.1", 30000, namespace));
+ return context;
+ }
+
+ @Test
+ void registrationRechecksChangesPublishedAfterTheInitialComparison()
throws Exception {
+ commit("1", "a", "changed-before-registration");
+ tasks.remove().run();
+ scheduler().submit(() -> { }).get(5, TimeUnit.SECONDS);
+ AsyncContext lateClient = client("a");
+ scheduler().submit(() -> { }).get(5, TimeUnit.SECONDS);
+ verify(lateClient).complete();
+ assertTrue(((MockHttpServletResponse)
lateClient.getResponse()).getContentAsString().contains("RULE"));
+ }
+
+ private ScheduledExecutorService scheduler() {
+ return (ScheduledExecutorService)
ReflectionTestUtils.getField(listener, "scheduler");
+ }
+
+ private void commit(final String id, final String namespace, final String
name) {
+ transactions.executeWithoutResult(status -> {
+ jdbc.update("INSERT INTO test_rule VALUES (?, ?, ?)", id,
namespace, name);
+ event(rule(id, namespace, name), DataEventTypeEnum.UPDATE);
+ });
+ }
+
+ private void event(final RuleData rule, final DataEventTypeEnum type) {
+ dispatcher.onApplicationEvent(new
DataChangedEvent(ConfigGroupEnum.RULE, type, List.of(rule)));
+ }
+
+ private static RuleData rule(final String id, final String namespace,
final String name) {
+ RuleData rule = new RuleData();
+ rule.setId(id);
+ rule.setNamespaceId(namespace);
+ rule.setName(name);
+ return rule;
+ }
+
+ private static final class TestListener extends
HttpLongPollingDataChangedListener {
+
+ private final AtomicInteger serializations = new AtomicInteger();
+
+ private final List<DataEventTypeEnum> callbacks =
Collections.synchronizedList(new ArrayList<>());
+
+ private TestListener(final HttpSyncProperties properties) {
+ super(properties);
+ }
+
+ @Override
+ protected <T> void updateCache(final ConfigGroupEnum group, final
List<T> data, final String namespaceId) {
+ serializations.incrementAndGet();
+ super.updateCache(group, data, namespaceId);
+ }
+
+ @Override
+ protected void afterRuleChanged(final List<RuleData> changed, final
DataEventTypeEnum eventType, final String namespaceId) {
+ callbacks.add(eventType);
+ super.afterRuleChanged(changed, eventType, namespaceId);
+ }
+
+ private ConfigDataCache cached(final String namespaceId) {
+ return CACHE.get(buildCacheKey(namespaceId,
ConfigGroupEnum.RULE.name()));
+ }
+
+ private void seedCache(final ConfigGroupEnum group, final String
namespaceId) {
+ String key = buildCacheKey(namespaceId, group.name());
+ CACHE.putIfAbsent(key, new ConfigDataCache(key, "[]", "seed", 0,
namespaceId));
+ }
+
+ private void clearCache() {
+ CACHE.clear();
+ }
+ }
+}
