You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@seata.apache.org by "YvCeung (via GitHub)" <gi...@apache.org> on 2025/12/25 16:39:45 UTC
[PR] feature:Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
YvCeung opened a new pull request, #7903:
URL: https://github.com/apache/incubator-seata/pull/7903
<!--
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.
-->
<!-- Please make sure you have read and understood the contributing guidelines -->
- [ ] I have read the [CONTRIBUTING.md](https://github.com/apache/incubator-seata/blob/2.x/CONTRIBUTING.md) guidelines.
- [ ] I have registered the PR [changes](https://github.com/apache/incubator-seata/tree/2.x/changes).
### Ⅰ. Describe what this PR did
### Ⅱ. Does this pull request fix one issue?
<!-- If that, add "fixes #xxx" below in the next line, for example, fixes #97. -->
### Ⅲ. Why don't you add test cases (unit test/integration test)?
### Ⅳ. Describe how to verify it
### Ⅴ. Special notes for reviews
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2735045941
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -89,36 +141,104 @@ public void init() {
@Async
public void onChangeEvent(ClusterChangeEvent event) {
if (event.getTerm() > 0) {
- GROUP_UPDATE_TERM.put(event.getGroup(), event.getTerm());
- // Notifications are made of changes in cluster information
- Optional.ofNullable(WATCHERS.remove(event.getGroup()))
- .ifPresent(watchers -> watchers.parallelStream().forEach(this::notifyWatcher));
+ String group = event.getGroup();
+ Long eventTerm = event.getTerm();
+
+ Long currentTerm = GROUP_UPDATE_TERM.get(group);
+ if (currentTerm != null && eventTerm <= currentTerm) {
+ logger.info(
+ "Discarding outdated event with term {} for group {}, current term is {}",
+ eventTerm,
+ group,
+ currentTerm);
+ return;
+ }
+
+ GROUP_UPDATE_TERM.put(group, eventTerm);
+
+ // Handle HTTP/1.1 watchers: remove and notify (one-time request)
+ Optional.ofNullable(HTTP1_WATCHERS.remove(group))
+ .ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
+ notifyWatcher(watcher, eventTerm);
+ // HTTP/1.1 watcher is done after notification
+ watcher.setDone(true);
+ }));
+
+ // Handle HTTP/2 watchers: notify without removing (long-lived connection)
+ Queue<Watcher<HttpContext>> http2Watchers = HTTP2_WATCHERS.get(group);
+ if (http2Watchers != null && !http2Watchers.isEmpty()) {
+ List<Watcher<HttpContext>> watchersToNotify = new ArrayList<>(http2Watchers);
+ watchersToNotify.forEach(watcher -> {
+ if (watcher.getAsyncContext().getContext().channel().isActive() && !watcher.isDone()) {
+ if (eventTerm > watcher.getTerm()) {
+ notifyWatcher(watcher, eventTerm);
+ } else {
+ logger.info(
+ "Skipping notification for watcher with term {} >= event term {} for group {}",
+ watcher.getTerm(),
+ eventTerm,
+ group);
+ }
+ } else {
+ // Remove inactive watcher
+ http2Watchers.remove(watcher);
+ HTTP2_HEADERS_SENT.remove(watcher);
+ }
+ });
+ }
}
}
- private void notifyWatcher(Watcher<HttpContext> watcher) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.OK);
+ private void notifyWatcher(Watcher<HttpContext> watcher, Long eventTerm) {
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context.isHttp2();
+
+ if (!isHttp2) {
+ watcher.setDone(true);
+ }
+
+ boolean isFirstResponse = !HTTP2_HEADERS_SENT.getOrDefault(watcher, false);
+ sendWatcherResponse(watcher, HttpResponseStatus.OK, false, isFirstResponse);
+ if (isFirstResponse && isHttp2) {
+ HTTP2_HEADERS_SENT.put(watcher, true);
+ }
+
+ if (eventTerm != null && eventTerm > watcher.getTerm()) {
+ watcher.setTerm(eventTerm);
+ }
}
+ /**
+ * Send watcher response to the client.
+ *
+ * @param watcher the watcher instance
+ * @param nettyStatus the HTTP status code
+ * @param closeStream whether to close the HTTP/2 stream (endStream=true)
+ * @param sendHeaders whether to send HTTP/2 headers frame (only needed for first response)
+ */
+ private void sendWatcherResponse(
+ Watcher<HttpContext> watcher,
+ HttpResponseStatus nettyStatus,
+ boolean closeStream,
+ boolean sendHeaders) {
- private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatus nettyStatus) {
- String group = watcher.getGroup();
HttpContext context = watcher.getAsyncContext();
if (!(context instanceof HttpContext)) {
logger.warn(
"Unsupported context type for watcher on group {}: {}",
- group,
+ watcher.getGroup(),
context != null ? context.getClass().getName() : "null");
return;
}
Review Comment:
The check on line 225 `if (!(context instanceof HttpContext))` is redundant since `context` is already declared as type `HttpContext` on line 224. This condition will always be false (unless there's some unexpected classloader issue). Consider removing this check or clarifying what the intended validation is.
##########
server/src/test/java/org/apache/seata/server/controller/ClusterControllerTest.java:
##########
@@ -131,70 +129,152 @@ public void run() {
throw new RuntimeException(e);
}
((ApplicationEventPublisher) ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT))
- .publishEvent(new ClusterChangeEvent(this, "default-test", 2, true));
+ .publishEvent(new ClusterChangeEvent(this, "default-test-group-2", 2, true));
}
});
thread.start();
try (Response response =
HttpClientUtil.doPost("http://127.0.0.1:" + port + "/metadata/v1/watch", param, header, 30000)) {
if (response != null) {
- Assertions.assertEquals(HttpStatus.SC_OK, response.code());
+ Assertions.assertEquals(200, response.code());
return;
}
}
Assertions.fail();
}
+ /**
+ * Verification points:
+ * 1. Verify HTTP/2 data push continuity: client can continuously receive events when server pushes them sequentially
+ * 2. Note: Cannot verify MetadataResponse due to inability to simulate real term changes in test environment
+ */
@Test
- @Order(5)
- void watch_withHttp2() throws Exception {
- CountDownLatch latch = new CountDownLatch(1);
+ @Order(4)
+ void watchStream_http2() throws Exception {
+ Map<String, String> header = new HashMap<>();
+ header.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
+ Map<String, String> param = new HashMap<>();
+ param.put("default-test-group-3", "1");
+
+ // Trigger a cluster change event after connection is established
+ Thread thread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
- Map<String, String> headers = new HashMap<>();
- headers.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
+ ((ApplicationEventPublisher) ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT))
+ .publishEvent(new ClusterChangeEvent(this, "default-test-group-3", 2, true));
+ }
+ });
+ thread.start();
- Map<String, String> params = new HashMap<>();
- params.put("default-test", "1");
+ try (SeataHttpWatch<ClusterWatchEvent> watch = HttpClientUtil.watchPost(
+ "http://127.0.0.1:" + port + "/metadata/v1/watch", param, header, ClusterWatchEvent.class)) {
+
+ // Verify HTTP/2 data push continuity: receive first event (connection established)
+ SeataHttpWatch.Response<ClusterWatchEvent> firstResponse = watch.next();
+ Assertions.assertNotNull(firstResponse.object, "First event data should not be null");
+ Assertions.assertEquals(SeataHttpWatch.Response.Type.UPDATE, firstResponse.type, "First event should be UPDATE");
+
+ // Verify HTTP/2 data push continuity: receive second event (cluster change event)
+ SeataHttpWatch.Response<ClusterWatchEvent> secondResponse = watch.next();
+ Assertions.assertNotNull(secondResponse.object, "Second event data should not be null");
+ Assertions.assertEquals(SeataHttpWatch.Response.Type.UPDATE, secondResponse.type, "Second event should be UPDATE");
+ Assertions.assertEquals("default-test-group-3", secondResponse.object.getGroup(), "Group should match");
+
+ logger.info("Successfully received two consecutive events from server");
+ }
+ }
+
+ /**
+ * Verification points:
+ * 1. Verify HTTP/2 data push continuity: client can continuously receive multiple events when server pushes them sequentially
+ * 2. Note: Cannot verify MetadataResponse due to inability to simulate real term changes in test environment
+ */
+ @Test
+ @Order(5)
+ void watchMultipleClusterUpdates_http2() throws Exception {
+ Map<String, String> header = new HashMap<>();
+ header.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
+ Map<String, String> param = new HashMap<>();
+ param.put("default-test-group-4", "1");
- Thread thread = new Thread(() -> {
+ // Trigger multiple cluster change events sequentially
+ Thread triggerThread = new Thread(() -> {
try {
- Thread.sleep(2000);
+ Thread.sleep(3000); // Wait for connection to be established
+
+ ((ApplicationEventPublisher) ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT))
+ .publishEvent(new ClusterChangeEvent(this, "default-test-group-4", 2, true));
+ Thread.sleep(1000);
+
+ ((ApplicationEventPublisher) ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT))
+ .publishEvent(new ClusterChangeEvent(this, "default-test-group-4", 3, true));
+ Thread.sleep(500);
+
+ ((ApplicationEventPublisher) ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT))
+ .publishEvent(new ClusterChangeEvent(this, "default-test-group-4", 4, true));
} catch (InterruptedException e) {
- throw new RuntimeException(e);
+ Thread.currentThread().interrupt();
}
- ((ApplicationEventPublisher) ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT))
- .publishEvent(new ClusterChangeEvent(this, "default-test", 2, true));
});
- thread.start();
+ triggerThread.start();
- HttpCallback<Response> callback = new HttpCallback<Response>() {
- @Override
- public void onSuccess(Response response) {
- Assertions.assertNotNull(response);
- Assertions.assertEquals(Protocol.H2_PRIOR_KNOWLEDGE, response.protocol());
- Assertions.assertEquals(HttpStatus.SC_OK, response.code());
- latch.countDown();
- }
+ boolean firstEventReceived = false;
+ int clusterUpdateCount = 0;
+ long startTime = System.currentTimeMillis();
+ long maxWaitTime = 10000;
+ int expectedUpdateCount = 3;
- @Override
- public void onFailure(Throwable t) {
- Assertions.fail("Should not fail: " + t.getMessage());
- }
+ // Verify HTTP/2 data push continuity: continuously receive multiple events
+ try (SeataHttpWatch<ClusterWatchEvent> watch = HttpClientUtil.watchPost(
+ "http://127.0.0.1:" + port + "/metadata/v1/watch", param, header, ClusterWatchEvent.class)) {
- @Override
- public void onCancelled() {
- Assertions.fail("Should not be cancelled");
+ while (System.currentTimeMillis() - startTime < maxWaitTime && clusterUpdateCount < expectedUpdateCount) {
+ try {
+ if (watch.hasNext()) {
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
+ if (response.type == SeataHttpWatch.Response.Type.UPDATE) {
+ Assertions.assertNotNull(response.object, "Event data should not be null");
+
+ if (!firstEventReceived) {
+ firstEventReceived = true;
+ logger.info("First event (connection established) received");
+ } else {
+ clusterUpdateCount++;
+ Assertions.assertEquals(
+ "default-test-group-4", response.object.getGroup(), "Group should match");
+ logger.info("Received cluster update event #{}", clusterUpdateCount);
+ }
Review Comment:
In the test on line 239, `watch.hasNext()` is called inside a while loop to check if there are more events. However, if the stream is blocked waiting for data (which is expected for an SSE connection), `hasNext()` might block or return false when there's no data yet available. This could lead to the loop exiting prematurely or spinning without actually waiting for events. Consider using `watch.next()` directly which will block until an event arrives, or implement a proper timeout mechanism that doesn't rely on `hasNext()`.
```suggestion
SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
if (response.type == SeataHttpWatch.Response.Type.UPDATE) {
Assertions.assertNotNull(response.object, "Event data should not be null");
if (!firstEventReceived) {
firstEventReceived = true;
logger.info("First event (connection established) received");
} else {
clusterUpdateCount++;
Assertions.assertEquals(
"default-test-group-4", response.object.getGroup(), "Group should match");
logger.info("Received cluster update event #{}", clusterUpdateCount);
```
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -128,31 +248,131 @@ private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatu
} else {
ctx.writeAndFlush(response);
}
- } else {
- // HTTP/2 response (h2c support)
- // Send headers frame
+ return;
+ }
+
+ // For HTTP/2, headers must be sent first on the initial response
+ if (sendHeaders) {
Http2Headers headers = new DefaultHttp2Headers().status(nettyStatus.codeAsText());
- headers.set(HttpHeaderNames.CONTENT_LENGTH, "0");
+ headers.set(HttpHeaderNames.CONTENT_TYPE, "text/event-stream; charset=utf-8");
+ headers.set(HttpHeaderNames.CACHE_CONTROL, "no-cache");
+
ctx.write(new DefaultHttp2HeadersFrame(headers));
+ }
- // Send empty data frame with endStream=true to close the stream
- ctx.writeAndFlush(new DefaultHttp2DataFrame(Unpooled.EMPTY_BUFFER, true))
- .addListener(f -> {
- if (!f.isSuccess()) {
- logger.warn("HTTP2 response send failed, group={}", group, f.cause());
- }
- });
+ String group = watcher.getGroup();
+ String eventData = buildEventData(group);
+ ByteBuf content = Unpooled.copiedBuffer(eventData, StandardCharsets.UTF_8);
+
+ // Send DATA frame (if closeStream is true, it will end the current stream)
+ ctx.write(new DefaultHttp2DataFrame(content, closeStream));
+ ctx.flush();
+ }
+
+ /**
+ * Get current cluster metadata for the given group.
+ * This method extracts the logic from ClusterController#cluster to avoid circular dependency.
+ *
+ * @param group the group name
+ * @return the MetadataResponse containing current cluster metadata
+ */
+ public MetadataResponse getMetadataResponse(String group) {
+ MetadataResponse metadataResponse = new MetadataResponse();
+ if (StringUtils.isBlank(group)) {
+ group = ConfigurationFactory.getInstance()
+ .getConfig(ConfigurationKeys.SERVER_RAFT_GROUP, DEFAULT_SEATA_GROUP);
+ }
+ RaftServer raftServer = RaftServerManager.getRaftServer(group);
+ if (raftServer != null) {
+ String mode = ConfigurationFactory.getInstance().getConfig(STORE_MODE);
+ metadataResponse.setStoreMode(mode);
+ RouteTable routeTable = RouteTable.getInstance();
+ try {
+ routeTable.refreshLeader(RaftServerManager.getCliClientServiceInstance(), group, 1000);
+ PeerId leader = routeTable.selectLeader(group);
+ if (leader != null) {
+ Set<Node> nodes = new HashSet<>();
+ RaftClusterMetadata raftClusterMetadata =
+ raftServer.getRaftStateMachine().getRaftLeaderMetadata();
+ Node leaderNode = raftServer
+ .getRaftStateMachine()
+ .getRaftLeaderMetadata()
+ .getLeader();
+ leaderNode.setGroup(group);
+ nodes.add(leaderNode);
+ nodes.addAll(raftClusterMetadata.getLearner());
+ nodes.addAll(raftClusterMetadata.getFollowers());
+ metadataResponse.setTerm(raftClusterMetadata.getTerm());
+ metadataResponse.setNodes(new ArrayList<>(nodes));
+ }
+ } catch (Exception e) {
+ logger.error("Failed to get cluster metadata for group {}: {}", group, e.getMessage(), e);
+ }
+ }
+ return metadataResponse;
+ }
+
+ /**
+ * Build event data with simplified format: only group, timestamp, and metadata.
+ * For HTTP/2 connections, this will send the full MetadataResponse.
+ *
+ * @param group the group name
+ * @return the event data string with prefix
+ */
+ private String buildEventData(String group) {
+ try {
+ // Get current MetadataResponse
+ MetadataResponse metadataResponse = getMetadataResponse(group);
+
+ // Build simplified JSON: only group, timestamp, and metadata
+ String json = String.format(
+ "{\"group\":\"%s\",\"timestamp\":%d,\"metadata\":%s}",
+ group,
+ System.currentTimeMillis(),
+ OBJECT_MAPPER.writeValueAsString(metadataResponse));
+
+ logger.debug("Sending watch event: group={}, term={}", group, metadataResponse.getTerm());
+ return Constants.WATCH_EVENT_PREFIX + json + "\n";
Review Comment:
The `buildEventData` method on line 328 uses `String.format` with `%s` to embed the JSON-serialized metadata directly into another JSON string. If the `group` parameter contains special characters like quotes or backslashes, this could result in malformed JSON. Consider properly escaping the `group` value or using a JSON builder/library to construct the entire JSON object safely rather than string formatting.
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -128,31 +248,131 @@ private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatu
} else {
ctx.writeAndFlush(response);
}
- } else {
- // HTTP/2 response (h2c support)
- // Send headers frame
+ return;
+ }
+
+ // For HTTP/2, headers must be sent first on the initial response
+ if (sendHeaders) {
Http2Headers headers = new DefaultHttp2Headers().status(nettyStatus.codeAsText());
- headers.set(HttpHeaderNames.CONTENT_LENGTH, "0");
+ headers.set(HttpHeaderNames.CONTENT_TYPE, "text/event-stream; charset=utf-8");
+ headers.set(HttpHeaderNames.CACHE_CONTROL, "no-cache");
+
ctx.write(new DefaultHttp2HeadersFrame(headers));
+ }
- // Send empty data frame with endStream=true to close the stream
- ctx.writeAndFlush(new DefaultHttp2DataFrame(Unpooled.EMPTY_BUFFER, true))
- .addListener(f -> {
- if (!f.isSuccess()) {
- logger.warn("HTTP2 response send failed, group={}", group, f.cause());
- }
- });
+ String group = watcher.getGroup();
+ String eventData = buildEventData(group);
+ ByteBuf content = Unpooled.copiedBuffer(eventData, StandardCharsets.UTF_8);
+
+ // Send DATA frame (if closeStream is true, it will end the current stream)
+ ctx.write(new DefaultHttp2DataFrame(content, closeStream));
+ ctx.flush();
Review Comment:
For HTTP/2 responses, the method sends event data by calling `buildEventData(group)` on line 264, which internally calls `getMetadataResponse(group)`. This operation involves Raft metadata lookups and could potentially fail or be expensive. If this fails or throws an exception after headers have been sent (line 260), the stream will be in an inconsistent state. Consider adding error handling to ensure that if `buildEventData` fails, an appropriate error response or stream closure is performed.
```suggestion
try {
String eventData = buildEventData(group);
ByteBuf content = Unpooled.copiedBuffer(eventData, StandardCharsets.UTF_8);
// Send DATA frame (if closeStream is true, it will end the current stream)
ctx.write(new DefaultHttp2DataFrame(content, closeStream));
ctx.flush();
} catch (Exception e) {
logger.error("Failed to build or send event data for watcher on group {}", group, e);
// Attempt to send an error event and close the stream to avoid leaving it in an inconsistent state
try {
String errorData = "event: error\ndata: internal server error\n\n";
ByteBuf errorContent = Unpooled.copiedBuffer(errorData, StandardCharsets.UTF_8);
ctx.write(new DefaultHttp2DataFrame(errorContent, true));
ctx.flush();
} catch (Exception inner) {
logger.error("Failed to send error response for watcher on group {}", group, inner);
}
}
```
##########
common/src/main/java/org/apache/seata/common/metadata/ClusterWatchEvent.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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.seata.common.metadata;
+
+/**
+ * Cluster watch event data class.
+ * Simplified format: only contains group, timestamp, and full metadata.
+ *
+ * <p>Event format:
+ * <pre>
+ * {"group":"default","timestamp":1234567890,"metadata":{"nodes":[...],"storeMode":"raft","term":2}}
+ * </pre>
+ *
+ * <p>Client can determine if update is needed by comparing metadata.term with local term.
+ * No need for separate event type field since all events contain full metadata.
+ *
+ * @see org.apache.seata.common.util.SeataHttpWatch
+ */
+public class ClusterWatchEvent {
+
+ private String group;
+
+ private Long timestamp;
+
+ private MetadataResponse metadata;
+
+ public ClusterWatchEvent() {}
+
+ public String getGroup() {
+ return group;
+ }
+
+ public void setGroup(String group) {
+ this.group = group;
+ }
+
+ public Long getTimestamp() {
+ return timestamp;
+ }
+
+ public void setTimestamp(Long timestamp) {
+ this.timestamp = timestamp;
+ }
+
+ public MetadataResponse getMetadata() {
+ return metadata;
+ }
+
+ public void setMetadata(MetadataResponse metadata) {
+ this.metadata = metadata;
+ }
+
+ @Override
+ public String toString() {
+ return "ClusterWatchEvent{" + "group='"
+ + group + '\'' + ", timestamp="
+ + timestamp + ", metadata="
+ + metadata + '}';
Review Comment:
Default toString(): MetadataResponse inherits toString() from Object, and so is not suitable for printing.
```suggestion
String metadataString = metadata == null
? "null"
: "MetadataResponse{term=" + metadata.getTerm() + "}";
return "ClusterWatchEvent{" + "group='"
+ group + '\'' + ", timestamp="
+ timestamp + ", metadata="
+ metadataString + '}';
```
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -128,31 +248,131 @@ private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatu
} else {
ctx.writeAndFlush(response);
}
- } else {
- // HTTP/2 response (h2c support)
- // Send headers frame
+ return;
+ }
+
+ // For HTTP/2, headers must be sent first on the initial response
+ if (sendHeaders) {
Http2Headers headers = new DefaultHttp2Headers().status(nettyStatus.codeAsText());
- headers.set(HttpHeaderNames.CONTENT_LENGTH, "0");
+ headers.set(HttpHeaderNames.CONTENT_TYPE, "text/event-stream; charset=utf-8");
+ headers.set(HttpHeaderNames.CACHE_CONTROL, "no-cache");
+
ctx.write(new DefaultHttp2HeadersFrame(headers));
+ }
- // Send empty data frame with endStream=true to close the stream
- ctx.writeAndFlush(new DefaultHttp2DataFrame(Unpooled.EMPTY_BUFFER, true))
- .addListener(f -> {
- if (!f.isSuccess()) {
- logger.warn("HTTP2 response send failed, group={}", group, f.cause());
- }
- });
+ String group = watcher.getGroup();
+ String eventData = buildEventData(group);
+ ByteBuf content = Unpooled.copiedBuffer(eventData, StandardCharsets.UTF_8);
+
+ // Send DATA frame (if closeStream is true, it will end the current stream)
+ ctx.write(new DefaultHttp2DataFrame(content, closeStream));
+ ctx.flush();
+ }
+
+ /**
+ * Get current cluster metadata for the given group.
+ * This method extracts the logic from ClusterController#cluster to avoid circular dependency.
+ *
+ * @param group the group name
+ * @return the MetadataResponse containing current cluster metadata
+ */
+ public MetadataResponse getMetadataResponse(String group) {
+ MetadataResponse metadataResponse = new MetadataResponse();
+ if (StringUtils.isBlank(group)) {
+ group = ConfigurationFactory.getInstance()
+ .getConfig(ConfigurationKeys.SERVER_RAFT_GROUP, DEFAULT_SEATA_GROUP);
+ }
+ RaftServer raftServer = RaftServerManager.getRaftServer(group);
+ if (raftServer != null) {
+ String mode = ConfigurationFactory.getInstance().getConfig(STORE_MODE);
+ metadataResponse.setStoreMode(mode);
+ RouteTable routeTable = RouteTable.getInstance();
+ try {
+ routeTable.refreshLeader(RaftServerManager.getCliClientServiceInstance(), group, 1000);
+ PeerId leader = routeTable.selectLeader(group);
+ if (leader != null) {
+ Set<Node> nodes = new HashSet<>();
+ RaftClusterMetadata raftClusterMetadata =
+ raftServer.getRaftStateMachine().getRaftLeaderMetadata();
+ Node leaderNode = raftServer
+ .getRaftStateMachine()
+ .getRaftLeaderMetadata()
+ .getLeader();
+ leaderNode.setGroup(group);
+ nodes.add(leaderNode);
+ nodes.addAll(raftClusterMetadata.getLearner());
+ nodes.addAll(raftClusterMetadata.getFollowers());
+ metadataResponse.setTerm(raftClusterMetadata.getTerm());
+ metadataResponse.setNodes(new ArrayList<>(nodes));
+ }
+ } catch (Exception e) {
+ logger.error("Failed to get cluster metadata for group {}: {}", group, e.getMessage(), e);
+ }
+ }
+ return metadataResponse;
+ }
+
+ /**
+ * Build event data with simplified format: only group, timestamp, and metadata.
+ * For HTTP/2 connections, this will send the full MetadataResponse.
+ *
+ * @param group the group name
+ * @return the event data string with prefix
+ */
+ private String buildEventData(String group) {
+ try {
+ // Get current MetadataResponse
+ MetadataResponse metadataResponse = getMetadataResponse(group);
+
+ // Build simplified JSON: only group, timestamp, and metadata
+ String json = String.format(
+ "{\"group\":\"%s\",\"timestamp\":%d,\"metadata\":%s}",
+ group,
+ System.currentTimeMillis(),
+ OBJECT_MAPPER.writeValueAsString(metadataResponse));
+
+ logger.debug("Sending watch event: group={}, term={}", group, metadataResponse.getTerm());
+ return Constants.WATCH_EVENT_PREFIX + json + "\n";
+ } catch (JsonProcessingException e) {
+ logger.error("Failed to serialize MetadataResponse for group {}: {}", group, e.getMessage(), e);
+ // Fallback: send minimal data
+ String json = String.format(
+ "{\"group\":\"%s\",\"timestamp\":%d,\"metadata\":null}",
+ group, System.currentTimeMillis());
+ return Constants.WATCH_EVENT_PREFIX + json + "\n";
}
}
public void registryWatcher(Watcher<HttpContext> watcher) {
String group = watcher.getGroup();
Long term = GROUP_UPDATE_TERM.get(group);
- if (term == null || watcher.getTerm() >= term) {
- WATCHERS.computeIfAbsent(group, value -> new ConcurrentLinkedQueue<>())
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context.isHttp2();
+
+ // For HTTP/2, must send response headers immediately, cannot delay
+ if (isHttp2 && !HTTP2_HEADERS_SENT.getOrDefault(watcher, false)) {
+ sendWatcherResponse(watcher, HttpResponseStatus.OK, false, true);
+ HTTP2_HEADERS_SENT.put(watcher, true);
+ }
+
+ if (isHttp2) {
+ HTTP2_WATCHERS
+ .computeIfAbsent(group, value -> new ConcurrentLinkedQueue<>())
.add(watcher);
+
+ // If term has been updated, notify immediately
+ if (term != null && term > watcher.getTerm()) {
+ notifyWatcher(watcher, null);
Review Comment:
There's a potential race condition between lines 353-356 in `registryWatcher` and lines 200-204 in `notifyWatcher`. If `registryWatcher` is called for an HTTP/2 watcher and simultaneously `notifyWatcher` is called (line 365), both might check `HTTP2_HEADERS_SENT.getOrDefault(watcher, false)` and find it false, causing headers to potentially be sent twice. While ConcurrentHashMap prevents data corruption, the logical race could result in duplicate header frames being sent. Consider using `computeIfAbsent` or synchronized blocks to ensure atomicity of the check-and-set operation.
##########
common/src/main/java/org/apache/seata/common/util/SeataHttpWatch.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.seata.common.util;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import okhttp3.Call;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.ResponseBody;
+import okio.BufferedSource;
+import org.apache.seata.common.Constants;
+import org.apache.seata.common.exception.FrameworkException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+/**
+ * Seata HTTP/2 Watch implementation.
+ * Consumes server-pushed event stream via an iterator-style API.
+ * Each line contains a single event in "data: {json}" format, similar to Kubernetes watch API.
+ *
+ * @param <T> event data type
+ */
+public class SeataHttpWatch<T>
+ implements Iterator<SeataHttpWatch.Response<T>>, Iterable<SeataHttpWatch.Response<T>>, AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SeataHttpWatch.class);
+
+ private final ResponseBody responseBody;
+ private final BufferedSource source;
+ private final Call call;
+ private final ObjectMapper objectMapper;
+ private final Class<T> eventType;
+
+ /**
+ * Response wrapper containing event data.
+ * Since event format is simplified (no type field), all successful events are treated the same.
+ * Client can determine the nature of the event by comparing metadata.term.
+ */
+ public static class Response<T> {
+ /**
+ * Response type enum
+ */
+ public enum Type {
+ /**
+ * Normal event with data (connection established or cluster changed)
+ */
+ UPDATE,
+ /**
+ * Error event (parse error or connection error)
+ */
+ ERROR
+ }
+
+ public final Type type;
+
+ public final T object;
+
+ public Response(Type type, T object) {
+ this.type = type;
+ this.object = object;
+ }
+ }
+
+ /**
+ * Create a Watch instance from an OkHttp call
+ *
+ * @param call the prepared HTTP call
+ * @param eventType the class type for deserializing event data
+ * @param <T> the event data type
+ * @return a Watch instance
+ * @throws IOException if the request fails
+ */
+ public static <T> SeataHttpWatch<T> createWatch(Call call, Class<T> eventType) throws IOException {
+
+ okhttp3.Response response = call.execute();
+
+ if (!response.isSuccessful()) {
+ String respBody = null;
+ try (ResponseBody body = response.body()) {
+ if (body != null) {
+ respBody = body.string();
+ }
+ } catch (IOException e) {
+ throw new FrameworkException(e, "Watch request failed: " + response.message());
+ }
+ throw new FrameworkException(
+ String.format("Watch request failed with code %d: %s", response.code(), respBody));
+ }
+
+ // Verify Content-Type is event stream
+ String contentType = response.header("Content-Type");
+ if (contentType == null || !contentType.contains("text/event-stream")) {
+ LOGGER.warn("Expected Content-Type: text/event-stream, got: {}", contentType);
+ }
+
+ return new SeataHttpWatch<>(response.body(), call, eventType);
Review Comment:
The `createWatch` method on line 90 executes the HTTP call and creates a SeataHttpWatch instance with the response body. However, if an exception occurs after line 92 but before line 113, the response object is never closed, leading to a resource leak. The response should be closed in a finally block or using try-with-resources to ensure proper cleanup, especially since the response.body() is passed to the SeataHttpWatch constructor.
```suggestion
boolean watchCreated = false;
try {
if (!response.isSuccessful()) {
String respBody = null;
try (ResponseBody body = response.body()) {
if (body != null) {
respBody = body.string();
}
} catch (IOException e) {
throw new FrameworkException(e, "Watch request failed: " + response.message());
}
throw new FrameworkException(
String.format("Watch request failed with code %d: %s", response.code(), respBody));
}
// Verify Content-Type is event stream
String contentType = response.header("Content-Type");
if (contentType == null || !contentType.contains("text/event-stream")) {
LOGGER.warn("Expected Content-Type: text/event-stream, got: {}", contentType);
}
SeataHttpWatch<T> watch = new SeataHttpWatch<>(response.body(), call, eventType);
watchCreated = true;
return watch;
} finally {
if (!watchCreated) {
ResponseBody body = response.body();
if (body != null) {
body.close();
} else {
response.close();
}
}
}
```
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -89,36 +141,104 @@ public void init() {
@Async
public void onChangeEvent(ClusterChangeEvent event) {
if (event.getTerm() > 0) {
- GROUP_UPDATE_TERM.put(event.getGroup(), event.getTerm());
- // Notifications are made of changes in cluster information
- Optional.ofNullable(WATCHERS.remove(event.getGroup()))
- .ifPresent(watchers -> watchers.parallelStream().forEach(this::notifyWatcher));
+ String group = event.getGroup();
+ Long eventTerm = event.getTerm();
+
+ Long currentTerm = GROUP_UPDATE_TERM.get(group);
+ if (currentTerm != null && eventTerm <= currentTerm) {
+ logger.info(
+ "Discarding outdated event with term {} for group {}, current term is {}",
+ eventTerm,
+ group,
+ currentTerm);
+ return;
+ }
+
+ GROUP_UPDATE_TERM.put(group, eventTerm);
+
+ // Handle HTTP/1.1 watchers: remove and notify (one-time request)
+ Optional.ofNullable(HTTP1_WATCHERS.remove(group))
+ .ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
+ notifyWatcher(watcher, eventTerm);
+ // HTTP/1.1 watcher is done after notification
+ watcher.setDone(true);
+ }));
+
+ // Handle HTTP/2 watchers: notify without removing (long-lived connection)
+ Queue<Watcher<HttpContext>> http2Watchers = HTTP2_WATCHERS.get(group);
+ if (http2Watchers != null && !http2Watchers.isEmpty()) {
+ List<Watcher<HttpContext>> watchersToNotify = new ArrayList<>(http2Watchers);
+ watchersToNotify.forEach(watcher -> {
+ if (watcher.getAsyncContext().getContext().channel().isActive() && !watcher.isDone()) {
+ if (eventTerm > watcher.getTerm()) {
+ notifyWatcher(watcher, eventTerm);
+ } else {
+ logger.info(
+ "Skipping notification for watcher with term {} >= event term {} for group {}",
+ watcher.getTerm(),
+ eventTerm,
+ group);
+ }
Review Comment:
The logger.info statement on lines 176-180 uses a positional argument format but the condition on line 173 compares `eventTerm > watcher.getTerm()`, meaning this log will only execute when the condition is false (i.e., when `eventTerm <= watcher.getTerm()`). However, the log message says "watcher with term {} >= event term {}", which would be confusing. The message should be adjusted to clarify it's logging when the notification is skipped because the watcher already has an equal or newer term.
##########
common/src/main/java/org/apache/seata/common/util/HttpClientUtil.java:
##########
@@ -265,25 +215,101 @@ private static String buildUrlWithParams(String url, Map<String, String> params)
return urlBuilder.toString();
}
- private static void executeAsync(OkHttpClient client, Request request, final HttpCallback<Response> callback) {
- client.newCall(request).enqueue(new Callback() {
- @Override
- public void onResponse(Call call, Response response) {
- try {
- callback.onSuccess(response);
- } finally {
- response.close();
- }
- }
+ /**
+ * Create an HTTP/2 client for watch connections.
+ * This client is configured for long-lived connections to receive Server-Sent Events (SSE).
+ * The client instances are cached and reused based on the connection timeout to improve performance.
+ *
+ * @param connectTimeoutSeconds connection timeout in seconds (fast failure if server is unreachable)
+ * @return configured OkHttpClient instance (cached and reused)
+ */
+ private static OkHttpClient createHttp2WatchClient(int connectTimeoutSeconds) {
+ return HTTP2_CLIENT_MAP.computeIfAbsent(connectTimeoutSeconds, k -> new OkHttpClient.Builder()
+ .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+ // Fast failure during connection phase
+ .connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ // Infinite read timeout to allow continuous listening for server push
+ .readTimeout(0, TimeUnit.SECONDS)
+ .writeTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ .build());
Review Comment:
The HTTP2_CLIENT_MAP uses `connectTimeoutSeconds` as the key (line 227), but the only place this is called is with the constant `HTTP2_WATCH_CONNECT_TIMEOUT_SECONDS` which is 30 seconds. This means there will only ever be one entry in this map. Consider whether the caching logic with a map is necessary, or if a simple static field would be more appropriate. The comment on line 221 mentions "cached and reused based on the connection timeout" but in practice, the timeout never varies.
```suggestion
* Singleton HTTP/2 client for watch connections.
* This client is configured for long-lived connections to receive Server-Sent Events (SSE)
* and is cached and reused to avoid repeatedly creating identical clients.
*/
private static volatile OkHttpClient HTTP2_WATCH_CLIENT;
/**
* Create (on first use) or retrieve the HTTP/2 client for watch connections.
* <p>
* The first invocation determines the connection timeout used to configure the client; subsequent
* calls return the same cached instance, regardless of the timeout value passed in.
*
* @param connectTimeoutSeconds connection timeout in seconds (fast failure if server is unreachable)
* @return configured OkHttpClient instance (cached and reused)
*/
private static OkHttpClient createHttp2WatchClient(int connectTimeoutSeconds) {
OkHttpClient client = HTTP2_WATCH_CLIENT;
if (client == null) {
synchronized (HttpClientUtil.class) {
client = HTTP2_WATCH_CLIENT;
if (client == null) {
client = new OkHttpClient.Builder()
.protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
// Fast failure during connection phase
.connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
// Infinite read timeout to allow continuous listening for server push
.readTimeout(0, TimeUnit.SECONDS)
.writeTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
.build();
HTTP2_WATCH_CLIENT = client;
}
}
}
return client;
```
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -89,36 +141,104 @@ public void init() {
@Async
public void onChangeEvent(ClusterChangeEvent event) {
if (event.getTerm() > 0) {
- GROUP_UPDATE_TERM.put(event.getGroup(), event.getTerm());
- // Notifications are made of changes in cluster information
- Optional.ofNullable(WATCHERS.remove(event.getGroup()))
- .ifPresent(watchers -> watchers.parallelStream().forEach(this::notifyWatcher));
+ String group = event.getGroup();
+ Long eventTerm = event.getTerm();
Review Comment:
The variable 'eventTerm' is only assigned values of primitive type and is never 'null', but it is declared with the boxed type 'Long'.
```suggestion
long eventTerm = event.getTerm();
```
##########
common/src/main/java/org/apache/seata/common/util/SeataHttpWatch.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.seata.common.util;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import okhttp3.Call;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.ResponseBody;
+import okio.BufferedSource;
+import org.apache.seata.common.Constants;
+import org.apache.seata.common.exception.FrameworkException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+/**
+ * Seata HTTP/2 Watch implementation.
+ * Consumes server-pushed event stream via an iterator-style API.
+ * Each line contains a single event in "data: {json}" format, similar to Kubernetes watch API.
+ *
+ * @param <T> event data type
+ */
+public class SeataHttpWatch<T>
+ implements Iterator<SeataHttpWatch.Response<T>>, Iterable<SeataHttpWatch.Response<T>>, AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SeataHttpWatch.class);
+
+ private final ResponseBody responseBody;
+ private final BufferedSource source;
+ private final Call call;
+ private final ObjectMapper objectMapper;
+ private final Class<T> eventType;
+
+ /**
+ * Response wrapper containing event data.
+ * Since event format is simplified (no type field), all successful events are treated the same.
+ * Client can determine the nature of the event by comparing metadata.term.
+ */
+ public static class Response<T> {
+ /**
+ * Response type enum
+ */
+ public enum Type {
+ /**
+ * Normal event with data (connection established or cluster changed)
+ */
+ UPDATE,
+ /**
+ * Error event (parse error or connection error)
+ */
+ ERROR
+ }
+
+ public final Type type;
+
+ public final T object;
+
+ public Response(Type type, T object) {
+ this.type = type;
+ this.object = object;
+ }
+ }
+
+ /**
+ * Create a Watch instance from an OkHttp call
+ *
+ * @param call the prepared HTTP call
+ * @param eventType the class type for deserializing event data
+ * @param <T> the event data type
+ * @return a Watch instance
+ * @throws IOException if the request fails
+ */
+ public static <T> SeataHttpWatch<T> createWatch(Call call, Class<T> eventType) throws IOException {
+
+ okhttp3.Response response = call.execute();
+
+ if (!response.isSuccessful()) {
+ String respBody = null;
+ try (ResponseBody body = response.body()) {
+ if (body != null) {
+ respBody = body.string();
+ }
+ } catch (IOException e) {
+ throw new FrameworkException(e, "Watch request failed: " + response.message());
+ }
+ throw new FrameworkException(
+ String.format("Watch request failed with code %d: %s", response.code(), respBody));
+ }
+
+ // Verify Content-Type is event stream
+ String contentType = response.header("Content-Type");
+ if (contentType == null || !contentType.contains("text/event-stream")) {
+ LOGGER.warn("Expected Content-Type: text/event-stream, got: {}", contentType);
+ }
+
+ return new SeataHttpWatch<>(response.body(), call, eventType);
+ }
+
+ /**
+ * Create a Watch instance with a prepared request
+ *
+ * @param client the OkHttpClient instance
+ * @param request the HTTP request
+ * @param eventType the class type for deserializing event data
+ * @param <T> the event data type
+ * @return a Watch instance
+ * @throws IOException if the request fails
+ */
+ public static <T> SeataHttpWatch<T> createWatch(OkHttpClient client, Request request, Class<T> eventType)
+ throws IOException {
+
+ Call call = client.newCall(request);
+ return createWatch(call, eventType);
+ }
+
+ private SeataHttpWatch(ResponseBody responseBody, Call call, Class<T> eventType) {
+ this.responseBody = responseBody;
+ this.source = responseBody.source();
+ this.call = call;
+ this.objectMapper = new ObjectMapper();
+ this.eventType = eventType;
+ }
+
+ @Override
+ public boolean hasNext() {
+ try {
+ // Check if source is exhausted (stream closed)
+ return !source.exhausted();
+ } catch (IOException e) {
+ LOGGER.error("Error checking if stream has more data", e);
+ return false;
+ }
+ }
+
+ @Override
+ public Response<T> next() {
+ try {
+ /*
+ Read a single line and parse it as an event.
+ Format: "{prefix}{json}\n" where prefix is defined in Constants.WATCH_EVENT_PREFIX.
+ Each line is a complete event, event type is included in the JSON data.
+ */
+ String line = source.readUtf8Line();
+ if (line == null) {
+ throw new RuntimeException("Stream closed unexpectedly");
+ }
+
+ if (!line.startsWith(Constants.WATCH_EVENT_PREFIX)) {
+ throw new RuntimeException("Invalid event format: expected prefix '" + Constants.WATCH_EVENT_PREFIX
+ + "', got: " + (line.length() > 20 ? line.substring(0, 20) + "..." : line));
+ }
+
+ String jsonData = line.substring(Constants.WATCH_EVENT_PREFIX.length());
+ return parseEvent(jsonData);
+
+ } catch (IOException e) {
+ throw new RuntimeException("IO Exception during next()", e);
+ }
+ }
+
+ /**
+ * Parse event JSON into Response object.
+ * Simplified format: only contains group, timestamp, and metadata fields.
+ *
+ * @param json the JSON string to parse
+ * @return the parsed Response object
+ * @throws IOException if parsing fails
+ */
+ private Response<T> parseEvent(String json) throws IOException {
+ try {
+ T eventData = objectMapper.readValue(json, eventType);
+ return new Response<>(Response.Type.UPDATE, eventData);
+
+ } catch (Exception e) {
+ LOGGER.error("Failed to parse event JSON: {}", json, e);
+ // Return error response
+ return new Response<>(Response.Type.ERROR, null);
+ }
Review Comment:
The `parseEvent` method catches a broad `Exception` on line 191, which includes both checked and unchecked exceptions. This could potentially catch and suppress serious errors like `OutOfMemoryError` or `StackOverflowError`. Consider catching only `JsonProcessingException` or `IOException` to handle parsing failures specifically, while allowing critical JVM errors to propagate.
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -89,36 +141,104 @@ public void init() {
@Async
public void onChangeEvent(ClusterChangeEvent event) {
if (event.getTerm() > 0) {
- GROUP_UPDATE_TERM.put(event.getGroup(), event.getTerm());
- // Notifications are made of changes in cluster information
- Optional.ofNullable(WATCHERS.remove(event.getGroup()))
- .ifPresent(watchers -> watchers.parallelStream().forEach(this::notifyWatcher));
+ String group = event.getGroup();
+ Long eventTerm = event.getTerm();
+
+ Long currentTerm = GROUP_UPDATE_TERM.get(group);
+ if (currentTerm != null && eventTerm <= currentTerm) {
+ logger.info(
+ "Discarding outdated event with term {} for group {}, current term is {}",
+ eventTerm,
+ group,
+ currentTerm);
+ return;
+ }
+
+ GROUP_UPDATE_TERM.put(group, eventTerm);
+
+ // Handle HTTP/1.1 watchers: remove and notify (one-time request)
+ Optional.ofNullable(HTTP1_WATCHERS.remove(group))
+ .ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
+ notifyWatcher(watcher, eventTerm);
+ // HTTP/1.1 watcher is done after notification
+ watcher.setDone(true);
+ }));
+
+ // Handle HTTP/2 watchers: notify without removing (long-lived connection)
+ Queue<Watcher<HttpContext>> http2Watchers = HTTP2_WATCHERS.get(group);
+ if (http2Watchers != null && !http2Watchers.isEmpty()) {
+ List<Watcher<HttpContext>> watchersToNotify = new ArrayList<>(http2Watchers);
+ watchersToNotify.forEach(watcher -> {
+ if (watcher.getAsyncContext().getContext().channel().isActive() && !watcher.isDone()) {
+ if (eventTerm > watcher.getTerm()) {
+ notifyWatcher(watcher, eventTerm);
+ } else {
+ logger.info(
+ "Skipping notification for watcher with term {} >= event term {} for group {}",
+ watcher.getTerm(),
+ eventTerm,
+ group);
+ }
+ } else {
+ // Remove inactive watcher
+ http2Watchers.remove(watcher);
+ HTTP2_HEADERS_SENT.remove(watcher);
+ }
+ });
Review Comment:
The iteration through `watchersToNotify` checks if watchers are active and not done, but on line 184, it attempts to remove from the original `http2Watchers` queue. Since `watchersToNotify` is a snapshot copy created on line 170, the `watcher` reference in the forEach on line 171 might not successfully remove the correct element from `http2Watchers` in all cases, especially if the queue has been modified concurrently. Consider tracking watchers to remove and performing a batch removal after the iteration, or using an iterator on the original queue with proper synchronization.
##########
common/src/main/java/org/apache/seata/common/util/SeataHttpWatch.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.seata.common.util;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import okhttp3.Call;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.ResponseBody;
+import okio.BufferedSource;
+import org.apache.seata.common.Constants;
+import org.apache.seata.common.exception.FrameworkException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+/**
+ * Seata HTTP/2 Watch implementation.
+ * Consumes server-pushed event stream via an iterator-style API.
+ * Each line contains a single event in "data: {json}" format, similar to Kubernetes watch API.
+ *
+ * @param <T> event data type
+ */
+public class SeataHttpWatch<T>
+ implements Iterator<SeataHttpWatch.Response<T>>, Iterable<SeataHttpWatch.Response<T>>, AutoCloseable {
Review Comment:
This Iterable is its own Iterator, but does not guard against multiple iterations.
```suggestion
implements Iterator<SeataHttpWatch.Response<T>>, AutoCloseable {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2777224468
##########
common/src/main/java/org/apache/seata/common/util/HttpClientUtil.java:
##########
@@ -265,25 +215,101 @@ private static String buildUrlWithParams(String url, Map<String, String> params)
return urlBuilder.toString();
}
- private static void executeAsync(OkHttpClient client, Request request, final HttpCallback<Response> callback) {
- client.newCall(request).enqueue(new Callback() {
- @Override
- public void onResponse(Call call, Response response) {
- try {
- callback.onSuccess(response);
- } finally {
- response.close();
- }
- }
+ /**
+ * Create an HTTP/2 client for watch connections.
+ * This client is configured for long-lived connections to receive Server-Sent Events (SSE).
+ * The client instances are cached and reused based on the connection timeout to improve performance.
+ *
+ * @param connectTimeoutSeconds connection timeout in seconds (fast failure if server is unreachable)
+ * @return configured OkHttpClient instance (cached and reused)
+ */
+ private static OkHttpClient createHttp2WatchClient(int connectTimeoutSeconds) {
+ return HTTP2_CLIENT_MAP.computeIfAbsent(connectTimeoutSeconds, k -> new OkHttpClient.Builder()
+ .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+ // Fast failure during connection phase
+ .connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ // Infinite read timeout to allow continuous listening for server push
+ .readTimeout(0, TimeUnit.SECONDS)
+ .writeTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ .build());
Review Comment:
Currently, read timeout is being used as the key, which is a more reasonable approach.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2659739519
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -90,35 +103,68 @@ public void init() {
public void onChangeEvent(ClusterChangeEvent event) {
if (event.getTerm() > 0) {
GROUP_UPDATE_TERM.put(event.getGroup(), event.getTerm());
- // Notifications are made of changes in cluster information
+ // Notify all watchers of cluster information changes
Optional.ofNullable(WATCHERS.remove(event.getGroup()))
.ifPresent(watchers -> watchers.parallelStream().forEach(this::notifyWatcher));
}
}
private void notifyWatcher(Watcher<HttpContext> watcher) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.OK);
- }
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
Review Comment:
done in
95b3ee6
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "funky-eyes (via GitHub)" <gi...@apache.org>.
funky-eyes commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2659252630
##########
common/src/main/java/org/apache/seata/common/util/HttpClientUtil.java:
##########
@@ -265,25 +215,101 @@ private static String buildUrlWithParams(String url, Map<String, String> params)
return urlBuilder.toString();
}
- private static void executeAsync(OkHttpClient client, Request request, final HttpCallback<Response> callback) {
- client.newCall(request).enqueue(new Callback() {
- @Override
- public void onResponse(Call call, Response response) {
- try {
- callback.onSuccess(response);
- } finally {
- response.close();
- }
- }
+ /**
+ * Create an HTTP/2 client for watch connections.
+ * This client is configured for long-lived connections to receive Server-Sent Events (SSE).
+ * The client instances are cached and reused based on the connection timeout to improve performance.
+ *
+ * @param connectTimeoutSeconds connection timeout in seconds (fast failure if server is unreachable)
+ * @return configured OkHttpClient instance (cached and reused)
+ */
+ private static OkHttpClient createHttp2WatchClient(int connectTimeoutSeconds) {
+ return HTTP2_CLIENT_MAP.computeIfAbsent(connectTimeoutSeconds, k -> new OkHttpClient.Builder()
+ .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+ // Fast failure during connection phase
+ .connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ // Infinite read timeout to allow continuous listening for server push
+ .readTimeout(0, TimeUnit.SECONDS)
Review Comment:
We should make the read timeout configurable instead of hardcoding it to 0. When readTimeout is set to 0, the underlying socket timeout becomes infinite, which can cause the request to block indefinitely. In cases where the server shuts down abruptly (e.g., a crash without a graceful TCP handshake), this may lead to permanent blocking.
The connectTimeout should be limited to around 10 seconds, while the readTimeout should be adjustable based on the specific request type.
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -57,24 +59,35 @@ public class ClusterWatcherManager implements ClusterChangeListener {
private static final Map<String, Long> GROUP_UPDATE_TERM = new ConcurrentHashMap<>();
+ private static final Map<Watcher<HttpContext>, Boolean> HTTP2_HEADERS_SENT = new ConcurrentHashMap<>();
+
private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor =
new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("long-polling", 1));
@PostConstruct
public void init() {
- // Responds to monitors that time out
+ // Periodically check and respond to watchers that have timed out
scheduledThreadPoolExecutor.scheduleAtFixedRate(
() -> {
for (String group : WATCHERS.keySet()) {
Optional.ofNullable(WATCHERS.remove(group))
.ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
- if (System.currentTimeMillis() >= watcher.getTimeout()) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.NOT_MODIFIED);
- }
- if (!watcher.isDone()) {
- // Re-register
- registryWatcher(watcher);
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
+ if (isHttp2) {
+ if (!context.getContext().channel().isActive()) {
+ watcher.setDone(true);
+ HTTP2_HEADERS_SENT.remove(watcher);
+ } else {
+ registryWatcher(watcher);
Review Comment:
If the cluster state changes again during the interval before re-registering for the subscription queue, is there a risk—in extreme cases—that the notification could be lost?
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -128,27 +174,63 @@ private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatu
} else {
ctx.writeAndFlush(response);
}
- } else {
- // HTTP/2 response (h2c support)
- // Send headers frame
+ return;
+ }
+
+ // For HTTP/2, headers must be sent first on the initial response
+ if (sendHeaders) {
Http2Headers headers = new DefaultHttp2Headers().status(nettyStatus.codeAsText());
- headers.set(HttpHeaderNames.CONTENT_LENGTH, "0");
+ headers.set(HttpHeaderNames.CONTENT_TYPE, "text/event-stream; charset=utf-8");
+ headers.set(HttpHeaderNames.CACHE_CONTROL, "no-cache");
+
ctx.write(new DefaultHttp2HeadersFrame(headers));
+ }
+
+ String group = watcher.getGroup();
+ String sse = buildSSEFormat(nettyStatus, closeStream, sendHeaders, group);
+
+ ByteBuf content = Unpooled.copiedBuffer(sse, StandardCharsets.UTF_8);
- // Send empty data frame with endStream=true to close the stream
- ctx.writeAndFlush(new DefaultHttp2DataFrame(Unpooled.EMPTY_BUFFER, true))
- .addListener(f -> {
- if (!f.isSuccess()) {
- logger.warn("HTTP2 response send failed, group={}", group, f.cause());
- }
- });
+ // Send DATA frame (if closeStream is true, it will end the current stream)
+ ctx.write(new DefaultHttp2DataFrame(content, closeStream));
+ ctx.flush();
+ }
+
+ private String buildSSEFormat(
+ HttpResponseStatus nettyStatus, boolean closeStream, boolean sendHeaders, String group) {
+ // Determine event type (embedded in JSON, not in SSE event field)
+ String eventType;
+ if (sendHeaders) {
+ // Send keepalive event when stream is first established to confirm connection
+ eventType = "keepalive";
+ } else if (closeStream && nettyStatus == HttpResponseStatus.NOT_MODIFIED) {
+ // Timeout event, stream needs to be closed
+ eventType = "timeout";
+ } else {
+ // Normal cluster update event
+ eventType = "cluster-update";
}
+
+ String json = String.format(
Review Comment:
I hope that when an HTTP/2 connection is established, the server directly pushes the cluster change results to the client, rather than requiring the client to initiate a separate request to fetch the cluster information.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2657054311
##########
common/src/main/java/org/apache/seata/common/util/HttpClientUtil.java:
##########
@@ -265,25 +188,77 @@ private static String buildUrlWithParams(String url, Map<String, String> params)
return urlBuilder.toString();
}
- private static void executeAsync(OkHttpClient client, Request request, final HttpCallback<Response> callback) {
- client.newCall(request).enqueue(new Callback() {
- @Override
- public void onResponse(Call call, Response response) {
- try {
- callback.onSuccess(response);
- } finally {
- response.close();
- }
- }
+ private static OkHttpClient createHttp2WatchClient(int connectTimeoutSeconds) {
+ return new OkHttpClient.Builder()
+ .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+ .connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS) // 连接阶段快速失败
+ .readTimeout(0, TimeUnit.SECONDS) // 等待TC推送数据(建立连接后持续监听服务器推送)
+ .writeTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ .build();
+ }
- @Override
- public void onFailure(Call call, IOException e) {
- if (call.isCanceled()) {
- callback.onCancelled();
- } else {
- callback.onFailure(e);
- }
- }
- });
+ public static <T> SeataHttpWatch<T> watch(String url, Map<String, String> headers, Class<T> eventType)
+ throws IOException {
+ return watch(url, headers, null, "GET", eventType);
+ }
+
+ public static <T> SeataHttpWatch<T> watch(String url, Class<T> eventType) throws IOException {
+ return watch(url, null, null, "GET", eventType);
+ }
Review Comment:
The non-empty url determination will be uniformly performed in the underlying watch method
```java
private static <T> SeataHttpWatch<T> watch(
String url, Map<String, String> headers, RequestBody requestBody, String method, Class<T> eventType)
throws IOException {
if (StringUtils.isBlank(url)) {
throw new IllegalArgumentException("URL must not be null or blank");
}
OkHttpClient client = createHttp2WatchClient(HTTP2_WATCH_CONNECT_TIMEOUT_SECONDS);
Request request = buildHttp2WatchRequest(url, headers, requestBody, method);
return SeataHttpWatch.createWatch(client, request, eventType);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#issuecomment-3705067733
@funky-eyes PTAL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2668253319
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -57,24 +59,35 @@ public class ClusterWatcherManager implements ClusterChangeListener {
private static final Map<String, Long> GROUP_UPDATE_TERM = new ConcurrentHashMap<>();
+ private static final Map<Watcher<HttpContext>, Boolean> HTTP2_HEADERS_SENT = new ConcurrentHashMap<>();
+
private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor =
new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("long-polling", 1));
@PostConstruct
public void init() {
- // Responds to monitors that time out
+ // Periodically check and respond to watchers that have timed out
scheduledThreadPoolExecutor.scheduleAtFixedRate(
() -> {
for (String group : WATCHERS.keySet()) {
Optional.ofNullable(WATCHERS.remove(group))
.ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
- if (System.currentTimeMillis() >= watcher.getTimeout()) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.NOT_MODIFIED);
- }
- if (!watcher.isDone()) {
- // Re-register
- registryWatcher(watcher);
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
+ if (isHttp2) {
+ if (!context.getContext().channel().isActive()) {
+ watcher.setDone(true);
+ HTTP2_HEADERS_SENT.remove(watcher);
+ } else {
+ registryWatcher(watcher);
Review Comment:
> I believe that writing data to the ChannelHandlerContext in a concurrent manner does have this possibility, but it does not affect the data push. It's just that several lines of data are pushed at once. As long as the data sent at one time is complete, it's fine. Because the client parses one line at a time. As for the disorder issue of the term, the client can make its own judgment. Of course, the server can also make some appropriate optimizations.
>
> 我认为并发往ChannelHandlerContext里面write数据是存在这种可能性，但是不影响数据推送，无非一下子推送了好几行数据，服务端只要确保一次性发出去的数据是完整的就行，因为客户端是一行行解析的。至于term的乱序问题 客户端可以自己判断，当然服务端也可以适当的做一点优化
这一块并发写入不会破坏数据完整性，不需要完全顺序处理，保持并发性能，只做必要的过滤优化 @funky-eyes wdyt?
```java
// For HTTP/2, headers must be sent first on the initial response
if (sendHeaders) {
Http2Headers headers = new DefaultHttp2Headers().status(nettyStatus.codeAsText());
headers.set(HttpHeaderNames.CONTENT_TYPE, "text/event-stream; charset=utf-8");
headers.set(HttpHeaderNames.CACHE_CONTROL, "no-cache");
ctx.write(new DefaultHttp2HeadersFrame(headers));
}
String group = watcher.getGroup();
Long finalTerm = term != null ? term : GROUP_UPDATE_TERM.getOrDefault(group, 0L);
String eventData = buildEventData(nettyStatus, closeStream, sendHeaders, group, finalTerm);
ByteBuf content = Unpooled.copiedBuffer(eventData, StandardCharsets.UTF_8);
// Send DATA frame (if closeStream is true, it will end the current stream)
ctx.write(new DefaultHttp2DataFrame(content, closeStream));
ctx.flush();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "funky-eyes (via GitHub)" <gi...@apache.org>.
funky-eyes commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2671624239
##########
common/src/main/java/org/apache/seata/common/util/SeataHttpWatch.java:
##########
Review Comment:
Additionally, I don't seem to see any usage of SeataHttpWatch on the client side.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2659612531
##########
common/src/main/java/org/apache/seata/common/util/SeataHttpWatch.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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.seata.common.util;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import okhttp3.Call;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.ResponseBody;
+import okio.BufferedSource;
+import org.apache.seata.common.exception.FrameworkException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+/**
+ * Seata HTTP/2 Watch implementation.
+ * Consumes server-pushed SSE data frames via an iterator-style API.
+ *
+ * @param <T> event data type
+ */
+public class SeataHttpWatch<T>
+ implements Iterator<SeataHttpWatch.Response<T>>, Iterable<SeataHttpWatch.Response<T>>, AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SeataHttpWatch.class);
+
+ private final ResponseBody responseBody;
+ private final BufferedSource source;
+ private final Call call;
+ private final ObjectMapper objectMapper;
+ private final Class<T> eventType;
+
+ /**
+ * Response wrapper containing event type and data
+ */
+ public static class Response<T> {
+ /**
+ * Event type enum - matches SSE event field values
+ */
+ public enum Type {
+ /**
+ * Cluster update event
+ */
+ CLUSTER_UPDATE,
+ /**
+ * Keep-alive event (connection established)
+ */
+ KEEPALIVE,
+ /**
+ * Timeout event (stream closed due to timeout)
+ */
+ TIMEOUT,
+ /**
+ * Error event
+ */
+ ERROR
+ }
+
+ public final Type type;
+
+ public final T object;
+
+ public Response(Type type, T object) {
+ this.type = type;
+ this.object = object;
+ }
+ }
+
+ /**
+ * Create a Watch instance from an OkHttp call
+ *
+ * @param call the prepared HTTP call
+ * @param eventType the class type for deserializing event data
+ * @param <T> the event data type
+ * @return a Watch instance
+ * @throws IOException if the request fails
+ */
+ public static <T> SeataHttpWatch<T> createWatch(Call call, Class<T> eventType) throws IOException {
+
+ okhttp3.Response response = call.execute();
+
+ if (!response.isSuccessful()) {
+ String respBody = null;
+ try (ResponseBody body = response.body()) {
+ if (body != null) {
+ respBody = body.string();
+ }
+ } catch (IOException e) {
+ throw new FrameworkException(e, "Watch request failed: " + response.message());
+ }
+ throw new FrameworkException(
+ String.format("Watch request failed with code %d: %s", response.code(), respBody));
+ }
+
+ // Verify Content-Type is SSE
+ String contentType = response.header("Content-Type");
+ if (contentType == null || !contentType.contains("text/event-stream")) {
+ LOGGER.warn("Expected Content-Type: text/event-stream, got: {}", contentType);
+ }
+
+ return new SeataHttpWatch<>(response.body(), call, eventType);
+ }
+
+ /**
+ * Create a Watch instance with a prepared request
+ *
+ * @param client the OkHttpClient instance
+ * @param request the HTTP request
+ * @param eventType the class type for deserializing event data
+ * @param <T> the event data type
+ * @return a Watch instance
+ * @throws IOException if the request fails
+ */
+ public static <T> SeataHttpWatch<T> createWatch(OkHttpClient client, Request request, Class<T> eventType)
+ throws IOException {
+
+ Call call = client.newCall(request);
+ return createWatch(call, eventType);
+ }
+
+ private SeataHttpWatch(ResponseBody responseBody, Call call, Class<T> eventType) {
+ this.responseBody = responseBody;
+ this.source = responseBody.source();
+ this.call = call;
+ this.objectMapper = new ObjectMapper();
+ this.eventType = eventType;
+ }
+
+ @Override
+ public boolean hasNext() {
+ try {
+ // Check if source is exhausted (stream closed)
+ return !source.exhausted();
+ } catch (IOException e) {
+ LOGGER.error("Error checking if stream has more data", e);
+ return false;
+ }
+ }
+
+ @Override
+ public Response<T> next() {
+ try {
+ // Read complete SSE event in a loop (not recursive to avoid stack overflow)
+ // SSE format: "data: {json}\n\n"
+ // Each event is separated by double newline
+ // Event type is included in the JSON data, not in a separate "event:" field
+
+ StringBuilder dataBuffer = new StringBuilder();
+
+ // Read lines until we get a complete event (ending with empty line)
+ // This loop reads all lines of a single event atomically
+ while (true) {
+ String line = source.readUtf8Line();
+
+ if (line == null) {
+ // Stream closed
+ if (dataBuffer.length() > 0) {
+ // We have partial data, try to parse it
+ String eventJson = dataBuffer.toString();
+ return parseEvent(eventJson);
+ }
+ throw new RuntimeException("Stream closed unexpectedly");
+ }
+
+ if (line.startsWith("data: ")) {
+ String jsonData = line.substring(6);
+ if (dataBuffer.length() > 0) {
+ dataBuffer.append('\n');
+ }
+ dataBuffer.append(jsonData);
+ } else if (line.isEmpty()) {
+ // Empty line indicates end of event
+ if (dataBuffer.length() > 0) {
+ String eventJson = dataBuffer.toString();
+ return parseEvent(eventJson);
+ }
+ // Empty line but no data, continue reading (skip blank lines between events)
+ } else {
+ LOGGER.debug("Unknown SSE line format, ignoring: {}", line);
+ }
+ }
+
+ } catch (IOException e) {
+ throw new RuntimeException("IO Exception during next()", e);
+ }
+ }
+
+ /**
+ * Parse SSE event JSON into Response object
+ *
+ * @param json the JSON string to parse (contains type field)
+ * @return the parsed Response object
+ * @throws IOException if parsing fails
+ */
+ private Response<T> parseEvent(String json) throws IOException {
+ try {
+ // Parse JSON once to get both event type and full data
+ // First, parse to extract event type
+ EventMetadata metadata = objectMapper.readValue(json, EventMetadata.class);
+
+ // Map event type from JSON to Response.Type
+ Response.Type responseType = mapEventTypeToResponseType(metadata.type);
+
+ // Deserialize the full event data (parse once, reuse if possible)
+ T eventData = objectMapper.readValue(json, eventType);
Review Comment:
done in [3e77c68](https://github.com/apache/incubator-seata/pull/7903/commits/3e77c681040a885ac861f82d5013b78155b246ad)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2665353392
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -90,35 +103,68 @@ public void init() {
public void onChangeEvent(ClusterChangeEvent event) {
if (event.getTerm() > 0) {
GROUP_UPDATE_TERM.put(event.getGroup(), event.getTerm());
- // Notifications are made of changes in cluster information
+ // Notify all watchers of cluster information changes
Optional.ofNullable(WATCHERS.remove(event.getGroup()))
.ifPresent(watchers -> watchers.parallelStream().forEach(this::notifyWatcher));
}
}
private void notifyWatcher(Watcher<HttpContext> watcher) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.OK);
- }
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context.isHttp2();
+
+ if (!isHttp2) {
+ watcher.setDone(true);
+ }
+
+ boolean isFirstResponse = !HTTP2_HEADERS_SENT.getOrDefault(watcher, false);
+ sendWatcherResponse(watcher, HttpResponseStatus.OK, false, isFirstResponse);
+ if (isFirstResponse && isHttp2) {
+ HTTP2_HEADERS_SENT.put(watcher, true);
+ }
- private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatus nettyStatus) {
+ // Update watcher's term to the latest term to prevent infinite loop
+ // This ensures that when registryWatcher is called, it won't trigger notifyWatcher again
String group = watcher.getGroup();
+ Long latestTerm = GROUP_UPDATE_TERM.get(group);
Review Comment:
done in [c5f13f7](https://github.com/apache/incubator-seata/pull/7903/commits/c5f13f709f7d7d3c46ba645532f57a28d78b5bcb)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2777303366
##########
common/src/main/java/org/apache/seata/common/util/SeataHttpWatch.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.seata.common.util;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import okhttp3.Call;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.ResponseBody;
+import okio.BufferedSource;
+import org.apache.seata.common.Constants;
+import org.apache.seata.common.exception.FrameworkException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+/**
+ * Seata HTTP/2 Watch implementation.
+ * Consumes server-pushed event stream via an iterator-style API.
+ * Each line contains a single event in "data: {json}" format, similar to Kubernetes watch API.
+ *
+ * @param <T> event data type
+ */
+public class SeataHttpWatch<T>
+ implements Iterator<SeataHttpWatch.Response<T>>, Iterable<SeataHttpWatch.Response<T>>, AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SeataHttpWatch.class);
+
+ private final ResponseBody responseBody;
+ private final BufferedSource source;
+ private final Call call;
+ private final ObjectMapper objectMapper;
+ private final Class<T> eventType;
+
+ /**
+ * Response wrapper containing event data.
+ * Since event format is simplified (no type field), all successful events are treated the same.
+ * Client can determine the nature of the event by comparing metadata.term.
+ */
+ public static class Response<T> {
+ /**
+ * Response type enum
+ */
+ public enum Type {
+ /**
+ * Normal event with data (connection established or cluster changed)
+ */
+ UPDATE,
+ /**
+ * Error event (parse error or connection error)
+ */
+ ERROR
+ }
+
+ public final Type type;
+
+ public final T object;
+
+ public Response(Type type, T object) {
+ this.type = type;
+ this.object = object;
+ }
+ }
+
+ /**
+ * Create a Watch instance from an OkHttp call
+ *
+ * @param call the prepared HTTP call
+ * @param eventType the class type for deserializing event data
+ * @param <T> the event data type
+ * @return a Watch instance
+ * @throws IOException if the request fails
+ */
+ public static <T> SeataHttpWatch<T> createWatch(Call call, Class<T> eventType) throws IOException {
+
+ okhttp3.Response response = call.execute();
+
+ if (!response.isSuccessful()) {
+ String respBody = null;
+ try (ResponseBody body = response.body()) {
+ if (body != null) {
+ respBody = body.string();
+ }
+ } catch (IOException e) {
+ throw new FrameworkException(e, "Watch request failed: " + response.message());
+ }
+ throw new FrameworkException(
+ String.format("Watch request failed with code %d: %s", response.code(), respBody));
+ }
+
+ // Verify Content-Type is event stream
+ String contentType = response.header("Content-Type");
+ if (contentType == null || !contentType.contains("text/event-stream")) {
+ LOGGER.warn("Expected Content-Type: text/event-stream, got: {}", contentType);
+ }
+
+ return new SeataHttpWatch<>(response.body(), call, eventType);
+ }
+
+ /**
+ * Create a Watch instance with a prepared request
+ *
+ * @param client the OkHttpClient instance
+ * @param request the HTTP request
+ * @param eventType the class type for deserializing event data
+ * @param <T> the event data type
+ * @return a Watch instance
+ * @throws IOException if the request fails
+ */
+ public static <T> SeataHttpWatch<T> createWatch(OkHttpClient client, Request request, Class<T> eventType)
+ throws IOException {
+
+ Call call = client.newCall(request);
+ return createWatch(call, eventType);
+ }
+
+ private SeataHttpWatch(ResponseBody responseBody, Call call, Class<T> eventType) {
+ this.responseBody = responseBody;
+ this.source = responseBody.source();
+ this.call = call;
+ this.objectMapper = new ObjectMapper();
+ this.eventType = eventType;
+ }
+
+ @Override
+ public boolean hasNext() {
+ try {
+ // Check if source is exhausted (stream closed)
+ return !source.exhausted();
+ } catch (IOException e) {
+ LOGGER.error("Error checking if stream has more data", e);
+ return false;
+ }
+ }
+
+ @Override
+ public Response<T> next() {
+ try {
+ /*
+ Read a single line and parse it as an event.
+ Format: "{prefix}{json}\n" where prefix is defined in Constants.WATCH_EVENT_PREFIX.
+ Each line is a complete event, event type is included in the JSON data.
+ */
+ String line = source.readUtf8Line();
+ if (line == null) {
+ throw new RuntimeException("Stream closed unexpectedly");
+ }
+
+ if (!line.startsWith(Constants.WATCH_EVENT_PREFIX)) {
+ throw new RuntimeException("Invalid event format: expected prefix '" + Constants.WATCH_EVENT_PREFIX
+ + "', got: " + (line.length() > 20 ? line.substring(0, 20) + "..." : line));
+ }
+
+ String jsonData = line.substring(Constants.WATCH_EVENT_PREFIX.length());
+ return parseEvent(jsonData);
+
+ } catch (IOException e) {
+ throw new RuntimeException("IO Exception during next()", e);
+ }
+ }
+
+ /**
+ * Parse event JSON into Response object.
+ * Simplified format: only contains group, timestamp, and metadata fields.
+ *
+ * @param json the JSON string to parse
+ * @return the parsed Response object
+ * @throws IOException if parsing fails
+ */
+ private Response<T> parseEvent(String json) throws IOException {
+ try {
+ T eventData = objectMapper.readValue(json, eventType);
+ return new Response<>(Response.Type.UPDATE, eventData);
+
+ } catch (Exception e) {
+ LOGGER.error("Failed to parse event JSON: {}", json, e);
+ // Return error response
+ return new Response<>(Response.Type.ERROR, null);
+ }
Review Comment:
This is unnecessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2657084549
##########
common/src/main/java/org/apache/seata/common/util/HttpClientUtil.java:
##########
@@ -265,25 +188,77 @@ private static String buildUrlWithParams(String url, Map<String, String> params)
return urlBuilder.toString();
}
- private static void executeAsync(OkHttpClient client, Request request, final HttpCallback<Response> callback) {
- client.newCall(request).enqueue(new Callback() {
- @Override
- public void onResponse(Call call, Response response) {
- try {
- callback.onSuccess(response);
- } finally {
- response.close();
- }
- }
+ private static OkHttpClient createHttp2WatchClient(int connectTimeoutSeconds) {
+ return new OkHttpClient.Builder()
+ .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+ .connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS) // 连接阶段快速失败
+ .readTimeout(0, TimeUnit.SECONDS) // 等待TC推送数据(建立连接后持续监听服务器推送)
+ .writeTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ .build();
+ }
- @Override
- public void onFailure(Call call, IOException e) {
- if (call.isCanceled()) {
- callback.onCancelled();
- } else {
- callback.onFailure(e);
- }
- }
- });
+ public static <T> SeataHttpWatch<T> watch(String url, Map<String, String> headers, Class<T> eventType)
+ throws IOException {
+ return watch(url, headers, null, "GET", eventType);
+ }
+
+ public static <T> SeataHttpWatch<T> watch(String url, Class<T> eventType) throws IOException {
+ return watch(url, null, null, "GET", eventType);
+ }
+
+ /**
+ * Execute a watch request with specified HTTP method and return a Watch iterator.
+ * This method creates a long-lived HTTP/2 connection to receive Server-Sent Events (SSE).
+ */
+ private static <T> SeataHttpWatch<T> watch(
+ String url, Map<String, String> headers, RequestBody requestBody, String method, Class<T> eventType)
+ throws IOException {
+
+ OkHttpClient client = createHttp2WatchClient(30);
+ Request request = buildHttp2WatchRequest(url, headers, requestBody, method);
+ return SeataHttpWatch.createWatch(client, request, eventType);
+ }
+
+ public static <T> SeataHttpWatch<T> watchPost(
+ String url, Map<String, String> params, Map<String, String> headers, Class<T> eventType)
+ throws IOException {
Review Comment:
The non-empty url determination will be uniformly performed in the underlying watch method
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2668236255
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -57,24 +59,35 @@ public class ClusterWatcherManager implements ClusterChangeListener {
private static final Map<String, Long> GROUP_UPDATE_TERM = new ConcurrentHashMap<>();
+ private static final Map<Watcher<HttpContext>, Boolean> HTTP2_HEADERS_SENT = new ConcurrentHashMap<>();
+
private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor =
new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("long-polling", 1));
@PostConstruct
public void init() {
- // Responds to monitors that time out
+ // Periodically check and respond to watchers that have timed out
scheduledThreadPoolExecutor.scheduleAtFixedRate(
() -> {
for (String group : WATCHERS.keySet()) {
Optional.ofNullable(WATCHERS.remove(group))
.ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
- if (System.currentTimeMillis() >= watcher.getTimeout()) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.NOT_MODIFIED);
- }
- if (!watcher.isDone()) {
- // Re-register
- registryWatcher(watcher);
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
+ if (isHttp2) {
+ if (!context.getContext().channel().isActive()) {
+ watcher.setDone(true);
+ HTTP2_HEADERS_SENT.remove(watcher);
+ } else {
+ registryWatcher(watcher);
Review Comment:
I believe that writing data to the ChannelHandlerContext in a concurrent manner does have this possibility, but it does not affect the data push. It's just that several lines of data are pushed at once. As long as the data sent at one time is complete, it's fine. Because the client parses one line at a time. As for the disorder issue of the term, the client can make its own judgment. Of course, the server can also make some appropriate optimizations.
我认为并发往ChannelHandlerContext里面write数据是存在这种可能性，但是不影响数据推送，无非一下子推送了好几行数据，服务端只要确保一次性发出去的数据是完整的就行，因为客户端是一行行解析的。至于term的乱序问题 客户端可以自己判断，当然服务端也可以适当的做一点优化
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2777190876
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -57,6 +59,8 @@ public class ClusterWatcherManager implements ClusterChangeListener {
private static final Map<String, Long> GROUP_UPDATE_TERM = new ConcurrentHashMap<>();
+ private static final Map<Watcher<HttpContext>, Boolean> HTTP2_HEADERS_SENT = new ConcurrentHashMap<>();
Review Comment:
There are already both proactive and passive cleanup mechanisms in place:
Proactive cleanup – A scheduled task runs every second, iterates over all HTTP/2 watchers, and removes entries from HTTP2_HEADERS_SENT when the corresponding channel is inactive.
Passive cleanup – Before sending a response, sendWatcherResponse checks whether the channel is still active; if not, it removes the watcher from HTTP2_HEADERS_SENT before returning.
So the map is cleared both by the periodic scan and by the pre-send check, not only when channels become inactive in an event-driven way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2777296972
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -128,31 +248,131 @@ private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatu
} else {
ctx.writeAndFlush(response);
}
- } else {
- // HTTP/2 response (h2c support)
- // Send headers frame
+ return;
+ }
+
+ // For HTTP/2, headers must be sent first on the initial response
+ if (sendHeaders) {
Http2Headers headers = new DefaultHttp2Headers().status(nettyStatus.codeAsText());
- headers.set(HttpHeaderNames.CONTENT_LENGTH, "0");
+ headers.set(HttpHeaderNames.CONTENT_TYPE, "text/event-stream; charset=utf-8");
+ headers.set(HttpHeaderNames.CACHE_CONTROL, "no-cache");
+
ctx.write(new DefaultHttp2HeadersFrame(headers));
+ }
- // Send empty data frame with endStream=true to close the stream
- ctx.writeAndFlush(new DefaultHttp2DataFrame(Unpooled.EMPTY_BUFFER, true))
- .addListener(f -> {
- if (!f.isSuccess()) {
- logger.warn("HTTP2 response send failed, group={}", group, f.cause());
- }
- });
+ String group = watcher.getGroup();
+ String eventData = buildEventData(group);
+ ByteBuf content = Unpooled.copiedBuffer(eventData, StandardCharsets.UTF_8);
+
+ // Send DATA frame (if closeStream is true, it will end the current stream)
+ ctx.write(new DefaultHttp2DataFrame(content, closeStream));
+ ctx.flush();
+ }
+
+ /**
+ * Get current cluster metadata for the given group.
+ * This method extracts the logic from ClusterController#cluster to avoid circular dependency.
+ *
+ * @param group the group name
+ * @return the MetadataResponse containing current cluster metadata
+ */
+ public MetadataResponse getMetadataResponse(String group) {
+ MetadataResponse metadataResponse = new MetadataResponse();
+ if (StringUtils.isBlank(group)) {
+ group = ConfigurationFactory.getInstance()
+ .getConfig(ConfigurationKeys.SERVER_RAFT_GROUP, DEFAULT_SEATA_GROUP);
+ }
+ RaftServer raftServer = RaftServerManager.getRaftServer(group);
+ if (raftServer != null) {
+ String mode = ConfigurationFactory.getInstance().getConfig(STORE_MODE);
+ metadataResponse.setStoreMode(mode);
+ RouteTable routeTable = RouteTable.getInstance();
+ try {
+ routeTable.refreshLeader(RaftServerManager.getCliClientServiceInstance(), group, 1000);
+ PeerId leader = routeTable.selectLeader(group);
+ if (leader != null) {
+ Set<Node> nodes = new HashSet<>();
+ RaftClusterMetadata raftClusterMetadata =
+ raftServer.getRaftStateMachine().getRaftLeaderMetadata();
+ Node leaderNode = raftServer
+ .getRaftStateMachine()
+ .getRaftLeaderMetadata()
+ .getLeader();
+ leaderNode.setGroup(group);
+ nodes.add(leaderNode);
+ nodes.addAll(raftClusterMetadata.getLearner());
+ nodes.addAll(raftClusterMetadata.getFollowers());
+ metadataResponse.setTerm(raftClusterMetadata.getTerm());
+ metadataResponse.setNodes(new ArrayList<>(nodes));
+ }
+ } catch (Exception e) {
+ logger.error("Failed to get cluster metadata for group {}: {}", group, e.getMessage(), e);
+ }
+ }
+ return metadataResponse;
+ }
+
+ /**
+ * Build event data with simplified format: only group, timestamp, and metadata.
+ * For HTTP/2 connections, this will send the full MetadataResponse.
+ *
+ * @param group the group name
+ * @return the event data string with prefix
+ */
+ private String buildEventData(String group) {
+ try {
+ // Get current MetadataResponse
+ MetadataResponse metadataResponse = getMetadataResponse(group);
+
+ // Build simplified JSON: only group, timestamp, and metadata
+ String json = String.format(
+ "{\"group\":\"%s\",\"timestamp\":%d,\"metadata\":%s}",
+ group,
+ System.currentTimeMillis(),
+ OBJECT_MAPPER.writeValueAsString(metadataResponse));
+
+ logger.debug("Sending watch event: group={}, term={}", group, metadataResponse.getTerm());
+ return Constants.WATCH_EVENT_PREFIX + json + "\n";
+ } catch (JsonProcessingException e) {
+ logger.error("Failed to serialize MetadataResponse for group {}: {}", group, e.getMessage(), e);
+ // Fallback: send minimal data
+ String json = String.format(
+ "{\"group\":\"%s\",\"timestamp\":%d,\"metadata\":null}",
+ group, System.currentTimeMillis());
+ return Constants.WATCH_EVENT_PREFIX + json + "\n";
}
}
public void registryWatcher(Watcher<HttpContext> watcher) {
String group = watcher.getGroup();
Long term = GROUP_UPDATE_TERM.get(group);
- if (term == null || watcher.getTerm() >= term) {
- WATCHERS.computeIfAbsent(group, value -> new ConcurrentLinkedQueue<>())
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context.isHttp2();
+
+ // For HTTP/2, must send response headers immediately, cannot delay
+ if (isHttp2 && !HTTP2_HEADERS_SENT.getOrDefault(watcher, false)) {
+ sendWatcherResponse(watcher, HttpResponseStatus.OK, false, true);
+ HTTP2_HEADERS_SENT.put(watcher, true);
+ }
+
+ if (isHttp2) {
+ HTTP2_WATCHERS
+ .computeIfAbsent(group, value -> new ConcurrentLinkedQueue<>())
.add(watcher);
+
+ // If term has been updated, notify immediately
+ if (term != null && term > watcher.getTerm()) {
+ notifyWatcher(watcher, null);
Review Comment:
在当前实现下不会出现该竞态，原因有两点：
执行顺序：registryWatcher 里是先执行 **发 headers + HTTP2_HEADERS_SENT.put(watcher, true)**（338–341 行），再把 **watcher 加入 HTTP2_WATCHERS**，之后才可能在同一调用内或由 onChangeEvent 触发 notifyWatcher。
调用关系：能执行到 notifyWatcher 的 watcher，要么来自同一次 registryWatcher（此时本线程已执行过 put），要么来自 onChangeEvent 遍历队列（此时该 watcher 已在队中，说明入队前已做过 put）。因此两处看到的都是「已发过 headers」，不会重复发送。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2659724142
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -57,24 +59,35 @@ public class ClusterWatcherManager implements ClusterChangeListener {
private static final Map<String, Long> GROUP_UPDATE_TERM = new ConcurrentHashMap<>();
+ private static final Map<Watcher<HttpContext>, Boolean> HTTP2_HEADERS_SENT = new ConcurrentHashMap<>();
+
private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor =
new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("long-polling", 1));
@PostConstruct
public void init() {
- // Responds to monitors that time out
+ // Periodically check and respond to watchers that have timed out
scheduledThreadPoolExecutor.scheduleAtFixedRate(
() -> {
for (String group : WATCHERS.keySet()) {
Optional.ofNullable(WATCHERS.remove(group))
.ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
- if (System.currentTimeMillis() >= watcher.getTimeout()) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.NOT_MODIFIED);
- }
- if (!watcher.isDone()) {
- // Re-register
- registryWatcher(watcher);
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
Review Comment:
done in [40c03f5](https://github.com/apache/incubator-seata/pull/7903/commits/40c03f5dd80ce5644ff7312c3f1e7588e504b15b)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2659727977
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -90,35 +103,68 @@ public void init() {
public void onChangeEvent(ClusterChangeEvent event) {
if (event.getTerm() > 0) {
GROUP_UPDATE_TERM.put(event.getGroup(), event.getTerm());
- // Notifications are made of changes in cluster information
+ // Notify all watchers of cluster information changes
Optional.ofNullable(WATCHERS.remove(event.getGroup()))
.ifPresent(watchers -> watchers.parallelStream().forEach(this::notifyWatcher));
}
}
private void notifyWatcher(Watcher<HttpContext> watcher) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.OK);
- }
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
+
+ if (!isHttp2) {
+ watcher.setDone(true);
+ }
+
+ boolean isFirstResponse = !HTTP2_HEADERS_SENT.getOrDefault(watcher, false);
+ sendWatcherResponse(watcher, HttpResponseStatus.OK, false, isFirstResponse);
+ if (isFirstResponse && isHttp2) {
+ HTTP2_HEADERS_SENT.put(watcher, true);
+ }
- private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatus nettyStatus) {
+ // Update watcher's term to the latest term to prevent infinite loop
+ // This ensures that when registryWatcher is called, it won't trigger notifyWatcher again
String group = watcher.getGroup();
+ Long latestTerm = GROUP_UPDATE_TERM.get(group);
+ if (latestTerm != null && latestTerm > watcher.getTerm()) {
+ watcher.setTerm(latestTerm);
+ }
+
+ // For HTTP/2, re-register the watcher to continue listening for future updates
+ if (isHttp2 && !watcher.isDone()) {
+ registryWatcher(watcher);
+ }
+ }
+ /**
+ * Send watcher response to the client.
+ *
+ * @param watcher the watcher instance
+ * @param nettyStatus the HTTP status code
+ * @param closeStream whether to close the HTTP/2 stream (endStream=true)
+ * @param sendHeaders whether to send HTTP/2 headers frame (only needed for first response)
+ */
+ private void sendWatcherResponse(
+ Watcher<HttpContext> watcher, HttpResponseStatus nettyStatus, boolean closeStream, boolean sendHeaders) {
+
HttpContext context = watcher.getAsyncContext();
if (!(context instanceof HttpContext)) {
logger.warn(
"Unsupported context type for watcher on group {}: {}",
- group,
+ watcher.getGroup(),
context != null ? context.getClass().getName() : "null");
return;
}
Review Comment:
I think type checking here is necessary, so the historical logic status quo has been retained, but the format of the printed logs has been modified
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "funky-eyes (via GitHub)" <gi...@apache.org>.
funky-eyes commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2757997034
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -128,31 +248,131 @@ private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatu
} else {
ctx.writeAndFlush(response);
}
- } else {
- // HTTP/2 response (h2c support)
- // Send headers frame
+ return;
+ }
+
+ // For HTTP/2, headers must be sent first on the initial response
+ if (sendHeaders) {
Http2Headers headers = new DefaultHttp2Headers().status(nettyStatus.codeAsText());
- headers.set(HttpHeaderNames.CONTENT_LENGTH, "0");
+ headers.set(HttpHeaderNames.CONTENT_TYPE, "text/event-stream; charset=utf-8");
+ headers.set(HttpHeaderNames.CACHE_CONTROL, "no-cache");
+
ctx.write(new DefaultHttp2HeadersFrame(headers));
+ }
- // Send empty data frame with endStream=true to close the stream
- ctx.writeAndFlush(new DefaultHttp2DataFrame(Unpooled.EMPTY_BUFFER, true))
- .addListener(f -> {
- if (!f.isSuccess()) {
- logger.warn("HTTP2 response send failed, group={}", group, f.cause());
- }
- });
+ String group = watcher.getGroup();
+ String eventData = buildEventData(group);
+ ByteBuf content = Unpooled.copiedBuffer(eventData, StandardCharsets.UTF_8);
+
+ // Send DATA frame (if closeStream is true, it will end the current stream)
+ ctx.write(new DefaultHttp2DataFrame(content, closeStream));
+ ctx.flush();
+ }
+
+ /**
+ * Get current cluster metadata for the given group.
+ * This method extracts the logic from ClusterController#cluster to avoid circular dependency.
+ *
+ * @param group the group name
+ * @return the MetadataResponse containing current cluster metadata
+ */
+ public MetadataResponse getMetadataResponse(String group) {
+ MetadataResponse metadataResponse = new MetadataResponse();
+ if (StringUtils.isBlank(group)) {
+ group = ConfigurationFactory.getInstance()
+ .getConfig(ConfigurationKeys.SERVER_RAFT_GROUP, DEFAULT_SEATA_GROUP);
+ }
+ RaftServer raftServer = RaftServerManager.getRaftServer(group);
+ if (raftServer != null) {
+ String mode = ConfigurationFactory.getInstance().getConfig(STORE_MODE);
+ metadataResponse.setStoreMode(mode);
+ RouteTable routeTable = RouteTable.getInstance();
+ try {
+ routeTable.refreshLeader(RaftServerManager.getCliClientServiceInstance(), group, 1000);
+ PeerId leader = routeTable.selectLeader(group);
Review Comment:
I believe these code changes are no longer necessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "funky-eyes (via GitHub)" <gi...@apache.org>.
funky-eyes commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2666816518
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -57,24 +59,35 @@ public class ClusterWatcherManager implements ClusterChangeListener {
private static final Map<String, Long> GROUP_UPDATE_TERM = new ConcurrentHashMap<>();
+ private static final Map<Watcher<HttpContext>, Boolean> HTTP2_HEADERS_SENT = new ConcurrentHashMap<>();
+
private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor =
new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("long-polling", 1));
@PostConstruct
public void init() {
- // Responds to monitors that time out
+ // Periodically check and respond to watchers that have timed out
scheduledThreadPoolExecutor.scheduleAtFixedRate(
() -> {
for (String group : WATCHERS.keySet()) {
Optional.ofNullable(WATCHERS.remove(group))
.ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
- if (System.currentTimeMillis() >= watcher.getTimeout()) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.NOT_MODIFIED);
- }
- if (!watcher.isDone()) {
- // Re-register
- registryWatcher(watcher);
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
+ if (isHttp2) {
+ if (!context.getContext().channel().isActive()) {
+ watcher.setDone(true);
+ HTTP2_HEADERS_SENT.remove(watcher);
+ } else {
+ registryWatcher(watcher);
Review Comment:
> Currently, optimizations have been made for potential loss issues. Different queues have been created for watching HTTP1 and HTTP2 respectively to store data. HTTP2 does not remove and re-register regularly. The scheduled tasks only check whether the channel is valid to prevent memory leaks.
>
> ```java
> // Separate watchers for HTTP/1.1 (one-time requests) and HTTP/2 (long-lived connections)
> private static final Map<String, Queue<Watcher<HttpContext>>> HTTP1_WATCHERS = new ConcurrentHashMap<>();
> private static final Map<String, Queue<Watcher<HttpContext>>> HTTP2_WATCHERS = new ConcurrentHashMap<>();
>
> // Check HTTP/2 watchers for connection validity (don't remove, just check)
> for (Map.Entry<String, Queue<Watcher<HttpContext>>> entry : HTTP2_WATCHERS.entrySet()) {
> String group = entry.getKey();
> Queue<Watcher<HttpContext>> watchers = entry.getValue();
> if (watchers == null || watchers.isEmpty()) {
> continue;
> }
>
> // Create snapshot to avoid concurrent modification
> List<Watcher<HttpContext>> watchersToCheck = new ArrayList<>(watchers);
>
> watchersToCheck.forEach(watcher -> {
> HttpContext context = watcher.getAsyncContext();
> if (!context.getContext().channel().isActive()) {
> // Remove invalid watcher
> watchers.remove(watcher);
> watcher.setDone(true);
> HTTP2_HEADERS_SENT.remove(watcher);
> logger.debug("Removed inactive HTTP/2 watcher for group: {}", group);
> }
> });
> }
> },
> ```
We also need to consider another issue: since the response is written back using HTTP/2 + SSE, is there any thread safety concern with concurrent writes to the SSE response? Additionally, could there be an out-of-order delivery problem—for example, if two events are generated, with event 1 arriving at the client after event 2, potentially causing the newer event to be overwritten by the older one?
To address this, should we ensure that events are delivered in order and include term validation? Specifically, if the term of a previous event in the response is already greater than that of a subsequent event, the latter should be discarded.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2665164922
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -57,24 +59,35 @@ public class ClusterWatcherManager implements ClusterChangeListener {
private static final Map<String, Long> GROUP_UPDATE_TERM = new ConcurrentHashMap<>();
+ private static final Map<Watcher<HttpContext>, Boolean> HTTP2_HEADERS_SENT = new ConcurrentHashMap<>();
+
private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor =
new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("long-polling", 1));
@PostConstruct
public void init() {
- // Responds to monitors that time out
+ // Periodically check and respond to watchers that have timed out
scheduledThreadPoolExecutor.scheduleAtFixedRate(
() -> {
for (String group : WATCHERS.keySet()) {
Optional.ofNullable(WATCHERS.remove(group))
.ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
- if (System.currentTimeMillis() >= watcher.getTimeout()) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.NOT_MODIFIED);
- }
- if (!watcher.isDone()) {
- // Re-register
- registryWatcher(watcher);
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
+ if (isHttp2) {
+ if (!context.getContext().channel().isActive()) {
+ watcher.setDone(true);
+ HTTP2_HEADERS_SENT.remove(watcher);
+ } else {
+ registryWatcher(watcher);
Review Comment:
Currently, optimizations have been made for potential loss issues. Different queues have been created for watching HTTP1 and HTTP2 respectively to store data. HTTP2 does not remove and re-register regularly. The scheduled tasks only check whether the channel is valid to prevent memory leaks.
```java
// Separate watchers for HTTP/1.1 (one-time requests) and HTTP/2 (long-lived connections)
private static final Map<String, Queue<Watcher<HttpContext>>> HTTP1_WATCHERS = new ConcurrentHashMap<>();
private static final Map<String, Queue<Watcher<HttpContext>>> HTTP2_WATCHERS = new ConcurrentHashMap<>();
// Check HTTP/2 watchers for connection validity (don't remove, just check)
for (Map.Entry<String, Queue<Watcher<HttpContext>>> entry : HTTP2_WATCHERS.entrySet()) {
String group = entry.getKey();
Queue<Watcher<HttpContext>> watchers = entry.getValue();
if (watchers == null || watchers.isEmpty()) {
continue;
}
// Create snapshot to avoid concurrent modification
List<Watcher<HttpContext>> watchersToCheck = new ArrayList<>(watchers);
watchersToCheck.forEach(watcher -> {
HttpContext context = watcher.getAsyncContext();
if (!context.getContext().channel().isActive()) {
// Remove invalid watcher
watchers.remove(watcher);
watcher.setDone(true);
HTTP2_HEADERS_SENT.remove(watcher);
logger.debug("Removed inactive HTTP/2 watcher for group: {}", group);
}
});
}
},
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#issuecomment-3784386435
@copilot
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature:Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2652265823
##########
server/src/test/java/org/apache/seata/server/controller/ClusterControllerTest.java:
##########
@@ -138,62 +139,84 @@ public void run() {
try (Response response =
HttpClientUtil.doPost("http://127.0.0.1:" + port + "/metadata/v1/watch", param, header, 30000)) {
if (response != null) {
- Assertions.assertEquals(HttpStatus.SC_OK, response.code());
+ Assertions.assertEquals(200, response.code());
return;
}
}
Assertions.fail();
}
@Test
- @Order(5)
- void watch_withHttp2() throws Exception {
- CountDownLatch latch = new CountDownLatch(1);
-
- Map<String, String> headers = new HashMap<>();
- headers.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
-
- Map<String, String> params = new HashMap<>();
- params.put("default-test", "1");
-
- Thread thread = new Thread(() -> {
- try {
- Thread.sleep(2000);
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
+ @Order(3)
+ void watch_stream() throws Exception {
+ Map<String, String> header = new HashMap<>();
+ header.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
+ Map<String, String> param = new HashMap<>();
+ param.put("default-test", "1");
+ Thread thread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ ((ApplicationEventPublisher) ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT))
+ .publishEvent(new ClusterChangeEvent(this, "default-test", 2, true));
}
- ((ApplicationEventPublisher) ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT))
- .publishEvent(new ClusterChangeEvent(this, "default-test", 2, true));
});
thread.start();
- HttpCallback<Response> callback = new HttpCallback<Response>() {
- @Override
- public void onSuccess(Response response) {
- Assertions.assertNotNull(response);
- Assertions.assertEquals(Protocol.H2_PRIOR_KNOWLEDGE, response.protocol());
- Assertions.assertEquals(HttpStatus.SC_OK, response.code());
- latch.countDown();
- }
-
- @Override
- public void onFailure(Throwable t) {
- Assertions.fail("Should not fail: " + t.getMessage());
+ boolean keepaliveReceived = false;
+ boolean clusterUpdateReceived = false;
+ long startTime = System.currentTimeMillis();
+ long maxWaitTime = 10000; // Maximum wait time: 10 seconds
+
+ try (SeataHttpWatch<ClusterWatchEvent> watch = HttpClientUtil.watchPost(
+ "http://127.0.0.1:" + port + "/metadata/v1/watch", param, header, ClusterWatchEvent.class)) {
+ // For HTTP2, connection will remain open, so we need to break after receiving expected events
+ while (watch.hasNext() && (System.currentTimeMillis() - startTime < maxWaitTime)) {
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
+ SeataHttpWatch.Response.Type type = response.type;
+
+ // 执行业务逻辑
Review Comment:
Chinese comment detected in test code. Comments should be in English for consistency with the rest of the codebase and to maintain accessibility for all contributors.
```suggestion
// Execute business logic
```
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -128,27 +180,64 @@ private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatu
} else {
ctx.writeAndFlush(response);
}
- } else {
- // HTTP/2 response (h2c support)
- // Send headers frame
+ return;
+ }
+
+ // 第一次响应，必须先发送 headers
+ if (sendHeaders) {
Http2Headers headers = new DefaultHttp2Headers().status(nettyStatus.codeAsText());
- headers.set(HttpHeaderNames.CONTENT_LENGTH, "0");
+ headers.set(HttpHeaderNames.CONTENT_TYPE, "text/event-stream; charset=utf-8");
+ headers.set(HttpHeaderNames.CACHE_CONTROL, "no-cache");
+
ctx.write(new DefaultHttp2HeadersFrame(headers));
+ }
+
+ String group = watcher.getGroup();
+ String sse = buildSSEFormat(nettyStatus, closeStream, sendHeaders, group);
+
+ ByteBuf content = Unpooled.copiedBuffer(sse, StandardCharsets.UTF_8);
- // Send empty data frame with endStream=true to close the stream
- ctx.writeAndFlush(new DefaultHttp2DataFrame(Unpooled.EMPTY_BUFFER, true))
- .addListener(f -> {
- if (!f.isSuccess()) {
- logger.warn("HTTP2 response send failed, group={}", group, f.cause());
- }
- });
+ // 发送 DATA 帧（closeStream = true 则结束本次 stream）
+ ctx.write(new DefaultHttp2DataFrame(content, closeStream));
+ ctx.flush();
+ }
+
+ private String buildSSEFormat(
+ HttpResponseStatus nettyStatus, boolean closeStream, boolean sendHeaders, String group) {
+ // 决定事件类型（放在 JSON 中，而不是 SSE event 字段）
+ String eventType;
+ if (sendHeaders) {
+ // 第一次建立 stream 时发送 keepalive 事件，确认连接建立
+ eventType = "keepalive";
+ } else if (closeStream && nettyStatus == HttpResponseStatus.NOT_MODIFIED) {
+ // 超时事件，需要关闭流
+ eventType = "timeout";
+ } else {
+ // 正常集群变更事件
+ eventType = "cluster-update";
}
+
+ // 构造 JSON 格式事件数据（包含 type 字段）
+ String json = String.format(
+ "{\"type\":\"%s\",\"group\":\"%s\",\"term\":%d,\"timestamp\":%d}",
+ eventType, group, GROUP_UPDATE_TERM.getOrDefault(group, 0L), System.currentTimeMillis());
+ logger.debug("Sending watch event: {}", json);
+
+ // SSE 格式：只发送 data: 字段，事件类型包含在 JSON 中
Review Comment:
Chinese comments detected in production code. Comments should be in English for consistency with the rest of the codebase and to maintain accessibility for all contributors.
```suggestion
// Determine the event type (encoded inside the JSON, not via the SSE "event" field)
String eventType;
if (sendHeaders) {
// On initial stream establishment, send a keepalive event to confirm the connection
eventType = "keepalive";
} else if (closeStream && nettyStatus == HttpResponseStatus.NOT_MODIFIED) {
// Timeout event: the stream should be closed
eventType = "timeout";
} else {
// Normal cluster change event
eventType = "cluster-update";
}
// Build JSON event payload (including the "type" field)
String json = String.format(
"{\"type\":\"%s\",\"group\":\"%s\",\"term\":%d,\"timestamp\":%d}",
eventType, group, GROUP_UPDATE_TERM.getOrDefault(group, 0L), System.currentTimeMillis());
logger.debug("Sending watch event: {}", json);
// SSE format: send only the data: field; the event type is carried inside the JSON
```
##########
common/src/main/java/org/apache/seata/common/util/HttpClientUtil.java:
##########
@@ -265,25 +188,77 @@ private static String buildUrlWithParams(String url, Map<String, String> params)
return urlBuilder.toString();
}
- private static void executeAsync(OkHttpClient client, Request request, final HttpCallback<Response> callback) {
- client.newCall(request).enqueue(new Callback() {
- @Override
- public void onResponse(Call call, Response response) {
- try {
- callback.onSuccess(response);
- } finally {
- response.close();
- }
- }
+ private static OkHttpClient createHttp2WatchClient(int connectTimeoutSeconds) {
+ return new OkHttpClient.Builder()
+ .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+ .connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS) // 连接阶段快速失败
+ .readTimeout(0, TimeUnit.SECONDS) // 等待TC推送数据(建立连接后持续监听服务器推送)
+ .writeTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ .build();
+ }
- @Override
- public void onFailure(Call call, IOException e) {
- if (call.isCanceled()) {
- callback.onCancelled();
- } else {
- callback.onFailure(e);
- }
- }
- });
+ public static <T> SeataHttpWatch<T> watch(String url, Map<String, String> headers, Class<T> eventType)
+ throws IOException {
+ return watch(url, headers, null, "GET", eventType);
+ }
+
+ public static <T> SeataHttpWatch<T> watch(String url, Class<T> eventType) throws IOException {
+ return watch(url, null, null, "GET", eventType);
+ }
+
+ /**
+ * Execute a watch request with specified HTTP method and return a Watch iterator.
+ * This method creates a long-lived HTTP/2 connection to receive Server-Sent Events (SSE).
+ */
+ private static <T> SeataHttpWatch<T> watch(
+ String url, Map<String, String> headers, RequestBody requestBody, String method, Class<T> eventType)
+ throws IOException {
+
+ OkHttpClient client = createHttp2WatchClient(30);
Review Comment:
Potential resource leak. The HTTP/2 client created by createHttp2WatchClient is not cached or reused, which could lead to resource exhaustion under high load. Consider implementing a client pool or caching mechanism similar to HTTP_CLIENT_MAP used for HTTP/1 clients.
##########
common/src/test/java/org/apache/seata/common/util/SeataHttpWatchTest.java:
##########
@@ -0,0 +1,790 @@
+/*
+ * 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.seata.common.util;
+
+import okhttp3.Call;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import okio.Buffer;
+import okio.BufferedSource;
+import org.apache.seata.common.exception.FrameworkException;
+import org.apache.seata.common.metadata.ClusterWatchEvent;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class SeataHttpWatchTest {
+
+ private OkHttpClient mockClient;
+ private Call mockCall;
+ private Response mockResponse;
+ private ResponseBody mockResponseBody;
+ private BufferedSource mockSource;
+
+ @BeforeEach
+ public void setUp() {
+ mockClient = mock(OkHttpClient.class);
+ mockCall = mock(Call.class);
+ mockResponse = mock(Response.class);
+ mockResponseBody = mock(ResponseBody.class);
+ mockSource = mock(BufferedSource.class);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ // Clean up if needed
+ }
+
+ @Test
+ public void testCreateWatch_WithSuccessfulResponse() throws IOException {
+ // Setup
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(mockSource);
+
+ // Execute
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Verify
+ assertNotNull(watch);
+ verify(mockClient).newCall(request);
+ verify(mockCall).execute();
+ }
+
+ @Test
+ public void testCreateWatch_WithUnsuccessfulResponse() throws IOException {
+ // Setup
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(false);
+ when(mockResponse.code()).thenReturn(404);
+ when(mockResponse.message()).thenReturn("Not Found");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.string()).thenReturn("Error message");
+
+ // Execute & Verify
+ FrameworkException exception = assertThrows(FrameworkException.class, () -> {
+ SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+ });
+
+ assertTrue(exception.getMessage().contains("404"));
+ assertTrue(exception.getMessage().contains("Error message"));
+ }
+
+ @Test
+ public void testCreateWatch_WithUnsuccessfulResponse_IOExceptionOnBodyRead() throws IOException {
+ // Setup
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(false);
+ when(mockResponse.code()).thenReturn(500);
+ when(mockResponse.message()).thenReturn("Internal Server Error");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.string()).thenThrow(new IOException("Read error"));
+
+ // Execute & Verify
+ FrameworkException exception = assertThrows(FrameworkException.class, () -> {
+ SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+ });
+
+ assertTrue(exception.getMessage().contains("Watch request failed"));
+ }
+
+ @Test
+ public void testCreateWatch_WithNullContentType() throws IOException {
+ // Setup
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn(null);
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(mockSource);
+
+ // Execute
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Verify - should still create watch but log warning
+ assertNotNull(watch);
+ }
+
+ @Test
+ public void testHasNext_WithAvailableData() throws IOException {
+ // Setup
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(mockSource);
+ when(mockSource.exhausted()).thenReturn(false);
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ boolean hasNext = watch.hasNext();
+
+ // Verify
+ assertTrue(hasNext);
+ verify(mockSource).exhausted();
+ }
+
+ @Test
+ public void testHasNext_WithNoMoreData() throws IOException {
+ // Setup
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(mockSource);
+ when(mockSource.exhausted()).thenReturn(true);
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ boolean hasNext = watch.hasNext();
+
+ // Verify
+ assertFalse(hasNext);
+ }
+
+ @Test
+ public void testHasNext_WithIOException() throws IOException {
+ // Setup
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(mockSource);
+ when(mockSource.exhausted()).thenThrow(new IOException("Stream error"));
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ boolean hasNext = watch.hasNext();
+
+ // Verify - should return false on IOException
+ assertFalse(hasNext);
+ }
+
+ @Test
+ public void testNext_WithKeepaliveEvent() throws IOException {
+ // Setup
+ String sseData = "data: {\"type\":\"keepalive\",\"group\":\"default-test\",\"timestamp\":1234567890}\n\n";
+ Buffer buffer = new Buffer();
+ buffer.writeString(sseData, StandardCharsets.UTF_8);
+
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(buffer);
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
+
+ // Verify
+ assertNotNull(response);
+ assertEquals(SeataHttpWatch.Response.Type.KEEPALIVE, response.type);
+ assertNotNull(response.object);
+ assertEquals("keepalive", response.object.getType());
+ assertEquals("default-test", response.object.getGroup());
+ assertEquals(1234567890L, response.object.getTimestamp());
+ }
+
+ @Test
+ public void testNext_WithClusterUpdateEvent() throws IOException {
+ // Setup
+ String sseData = "data: {\"type\":\"cluster-update\",\"group\":\"default-test\",\"term\":2,\"timestamp\":1234567890}\n\n";
+ Buffer buffer = new Buffer();
+ buffer.writeString(sseData, StandardCharsets.UTF_8);
+
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(buffer);
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
+
+ // Verify
+ assertNotNull(response);
+ assertEquals(SeataHttpWatch.Response.Type.CLUSTER_UPDATE, response.type);
+ assertNotNull(response.object);
+ assertEquals("cluster-update", response.object.getType());
+ assertEquals("default-test", response.object.getGroup());
+ assertEquals(2L, response.object.getTerm());
+ assertEquals(1234567890L, response.object.getTimestamp());
+ }
+
+ @Test
+ public void testNext_WithTimeoutEvent() throws IOException {
+ // Setup
+ String sseData = "data: {\"type\":\"timeout\",\"group\":\"default-test\",\"timestamp\":1234567890}\n\n";
+ Buffer buffer = new Buffer();
+ buffer.writeString(sseData, StandardCharsets.UTF_8);
+
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(buffer);
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
+
+ // Verify
+ assertNotNull(response);
+ assertEquals(SeataHttpWatch.Response.Type.TIMEOUT, response.type);
+ assertNotNull(response.object);
+ assertEquals("timeout", response.object.getType());
+ }
+
+ @Test
+ public void testNext_WithUnknownEventType() throws IOException {
+ // Setup
+ String sseData = "data: {\"type\":\"unknown-type\",\"group\":\"default-test\",\"timestamp\":1234567890}\n\n";
+ Buffer buffer = new Buffer();
+ buffer.writeString(sseData, StandardCharsets.UTF_8);
+
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(buffer);
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
+
+ // Verify - should default to CLUSTER_UPDATE for unknown types
+ assertNotNull(response);
+ assertEquals(SeataHttpWatch.Response.Type.CLUSTER_UPDATE, response.type);
+ }
+
+ @Test
+ public void testNext_WithNullEventType() throws IOException {
+ // Setup
+ String sseData = "data: {\"group\":\"default-test\",\"timestamp\":1234567890}\n\n";
+ Buffer buffer = new Buffer();
+ buffer.writeString(sseData, StandardCharsets.UTF_8);
+
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(buffer);
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
+
+ // Verify - should default to CLUSTER_UPDATE when type is null
+ assertNotNull(response);
+ assertEquals(SeataHttpWatch.Response.Type.CLUSTER_UPDATE, response.type);
+ }
+
+ @Test
+ public void testNext_WithInvalidJson() throws IOException {
+ // Setup
+ String sseData = "data: {invalid json}\n\n";
+ Buffer buffer = new Buffer();
+ buffer.writeString(sseData, StandardCharsets.UTF_8);
+
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(buffer);
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
+
+ // Verify - should return ERROR type with null object
+ assertNotNull(response);
+ assertEquals(SeataHttpWatch.Response.Type.ERROR, response.type);
+ assertNull(response.object);
+ }
+
+ // TC发送SSE的固定格式\n\n 一次事件只发送一行,客户端每次读一行
+ @Test
+ public void testNext_WithMultipleEvents() throws IOException {
+ // 模拟两个事件
Review Comment:
Chinese comment detected in test code. Comments should be in English for consistency with the rest of the codebase and to maintain accessibility for all contributors.
```suggestion
// TC sends SSE in a fixed format "\n\n": each event contains a single line and the client reads one line at a time
@Test
public void testNext_WithMultipleEvents() throws IOException {
// Simulate two events
```
##########
common/src/main/java/org/apache/seata/common/util/HttpClientUtil.java:
##########
@@ -265,25 +188,77 @@ private static String buildUrlWithParams(String url, Map<String, String> params)
return urlBuilder.toString();
}
- private static void executeAsync(OkHttpClient client, Request request, final HttpCallback<Response> callback) {
- client.newCall(request).enqueue(new Callback() {
- @Override
- public void onResponse(Call call, Response response) {
- try {
- callback.onSuccess(response);
- } finally {
- response.close();
- }
- }
+ private static OkHttpClient createHttp2WatchClient(int connectTimeoutSeconds) {
+ return new OkHttpClient.Builder()
+ .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+ .connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS) // 连接阶段快速失败
+ .readTimeout(0, TimeUnit.SECONDS) // 等待TC推送数据(建立连接后持续监听服务器推送)
+ .writeTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ .build();
+ }
- @Override
- public void onFailure(Call call, IOException e) {
- if (call.isCanceled()) {
- callback.onCancelled();
- } else {
- callback.onFailure(e);
- }
- }
- });
+ public static <T> SeataHttpWatch<T> watch(String url, Map<String, String> headers, Class<T> eventType)
+ throws IOException {
+ return watch(url, headers, null, "GET", eventType);
+ }
+
+ public static <T> SeataHttpWatch<T> watch(String url, Class<T> eventType) throws IOException {
+ return watch(url, null, null, "GET", eventType);
+ }
+
+ /**
+ * Execute a watch request with specified HTTP method and return a Watch iterator.
+ * This method creates a long-lived HTTP/2 connection to receive Server-Sent Events (SSE).
+ */
+ private static <T> SeataHttpWatch<T> watch(
+ String url, Map<String, String> headers, RequestBody requestBody, String method, Class<T> eventType)
+ throws IOException {
+
+ OkHttpClient client = createHttp2WatchClient(30);
Review Comment:
Magic number detected. The connection timeout value of 30 seconds should be extracted to a named constant with clear documentation explaining why this specific timeout is appropriate for watch connections.
##########
common/src/main/java/org/apache/seata/common/util/SeataHttpWatch.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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.seata.common.util;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import okhttp3.Call;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.ResponseBody;
+import okio.BufferedSource;
+import org.apache.seata.common.exception.FrameworkException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+/**
+ * Seata HTTP/2 Watch implementation.
+ * Consumes server-pushed SSE data frames via an iterator-style API.
+ *
+ * @param <T> event data type
+ */
+public class SeataHttpWatch<T>
+ implements Iterator<SeataHttpWatch.Response<T>>, Iterable<SeataHttpWatch.Response<T>>, AutoCloseable {
Review Comment:
This Iterable is its own Iterator, but does not guard against multiple iterations.
```suggestion
implements Iterator<SeataHttpWatch.Response<T>>, AutoCloseable {
```
##########
common/src/main/java/org/apache/seata/common/util/HttpClientUtil.java:
##########
@@ -265,25 +188,77 @@ private static String buildUrlWithParams(String url, Map<String, String> params)
return urlBuilder.toString();
}
- private static void executeAsync(OkHttpClient client, Request request, final HttpCallback<Response> callback) {
- client.newCall(request).enqueue(new Callback() {
- @Override
- public void onResponse(Call call, Response response) {
- try {
- callback.onSuccess(response);
- } finally {
- response.close();
- }
- }
+ private static OkHttpClient createHttp2WatchClient(int connectTimeoutSeconds) {
+ return new OkHttpClient.Builder()
+ .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+ .connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS) // 连接阶段快速失败
+ .readTimeout(0, TimeUnit.SECONDS) // 等待TC推送数据(建立连接后持续监听服务器推送)
+ .writeTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ .build();
+ }
- @Override
- public void onFailure(Call call, IOException e) {
- if (call.isCanceled()) {
- callback.onCancelled();
- } else {
- callback.onFailure(e);
- }
- }
- });
+ public static <T> SeataHttpWatch<T> watch(String url, Map<String, String> headers, Class<T> eventType)
+ throws IOException {
+ return watch(url, headers, null, "GET", eventType);
+ }
+
+ public static <T> SeataHttpWatch<T> watch(String url, Class<T> eventType) throws IOException {
+ return watch(url, null, null, "GET", eventType);
+ }
+
+ /**
+ * Execute a watch request with specified HTTP method and return a Watch iterator.
+ * This method creates a long-lived HTTP/2 connection to receive Server-Sent Events (SSE).
+ */
+ private static <T> SeataHttpWatch<T> watch(
+ String url, Map<String, String> headers, RequestBody requestBody, String method, Class<T> eventType)
+ throws IOException {
+
+ OkHttpClient client = createHttp2WatchClient(30);
+ Request request = buildHttp2WatchRequest(url, headers, requestBody, method);
+ return SeataHttpWatch.createWatch(client, request, eventType);
+ }
+
+ public static <T> SeataHttpWatch<T> watchPost(
+ String url, Map<String, String> params, Map<String, String> headers, Class<T> eventType)
+ throws IOException {
+ try {
+ String contentType = headers != null ? headers.get("Content-Type") : "";
+ RequestBody requestBody = createRequestBody(params, contentType);
+ return watch(url, headers, requestBody, "POST", eventType);
+ } catch (JsonProcessingException e) {
+ LOGGER.error("Failed to create request body: {}", e.getMessage(), e);
Review Comment:
Inconsistent error handling. The method wraps JsonProcessingException in IOException with a descriptive message, but doesn't preserve the cause properly for the logging statement. The logging on line 230 logs the message, then the code throws with the same message. Consider either logging with the full exception context or removing the redundant log statement.
```suggestion
LOGGER.error("Failed to create request body", e);
```
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -97,28 +115,62 @@ public void onChangeEvent(ClusterChangeEvent event) {
}
private void notifyWatcher(Watcher<HttpContext> watcher) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.OK);
- }
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
+
+ if (!isHttp2) {
+ watcher.setDone(true);
+ }
+
+ boolean isFirstResponse = !HTTP2_HEADERS_SENT.getOrDefault(watcher, false);
+ sendWatcherResponse(watcher, HttpResponseStatus.OK, false, isFirstResponse);
+ if (isFirstResponse && isHttp2) {
+ HTTP2_HEADERS_SENT.put(watcher, true);
+ }
- private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatus nettyStatus) {
+ // Update watcher's term to the latest term to prevent infinite loop
+ // This ensures that when registryWatcher is called, it won't trigger notifyWatcher again
String group = watcher.getGroup();
+ Long latestTerm = GROUP_UPDATE_TERM.get(group);
+ if (latestTerm != null && latestTerm > watcher.getTerm()) {
+ watcher.setTerm(latestTerm);
+ }
+
+ // For HTTP/2, re-register the watcher to continue listening for future updates
+ if (isHttp2 && !watcher.isDone()) {
+ registryWatcher(watcher);
+ }
+ }
+ /**
+ * Send watcher response to the client.
+ *
+ * @param watcher the watcher instance
+ * @param nettyStatus the HTTP status code
+ * @param closeStream whether to close the HTTP/2 stream (endStream=true)
+ * @param sendHeaders whether to send HTTP/2 headers frame (only needed for first response)
+ */
+ private void sendWatcherResponse(
+ Watcher<HttpContext> watcher, HttpResponseStatus nettyStatus, boolean closeStream, boolean sendHeaders) {
+
HttpContext context = watcher.getAsyncContext();
if (!(context instanceof HttpContext)) {
logger.warn(
"Unsupported context type for watcher on group {}: {}",
- group,
+ watcher.getGroup(),
context != null ? context.getClass().getName() : "null");
return;
}
ChannelHandlerContext ctx = context.getContext();
+
if (!ctx.channel().isActive()) {
- logger.warn("Netty channel is not active for watcher on group {}, cannot send response.", group);
+ HTTP2_HEADERS_SENT.remove(watcher);
+ logger.warn(
+ "Netty channel is not active for watcher on group {}, cannot send response.", watcher.getGroup());
return;
}
+ // HTTP/1 长轮询保持原逻辑
Review Comment:
Chinese comments detected in production code. Comments should be in English for consistency with the rest of the codebase and to maintain accessibility for all contributors.
##########
common/src/main/java/org/apache/seata/common/util/HttpClientUtil.java:
##########
@@ -265,25 +188,77 @@ private static String buildUrlWithParams(String url, Map<String, String> params)
return urlBuilder.toString();
}
- private static void executeAsync(OkHttpClient client, Request request, final HttpCallback<Response> callback) {
- client.newCall(request).enqueue(new Callback() {
- @Override
- public void onResponse(Call call, Response response) {
- try {
- callback.onSuccess(response);
- } finally {
- response.close();
- }
- }
+ private static OkHttpClient createHttp2WatchClient(int connectTimeoutSeconds) {
+ return new OkHttpClient.Builder()
+ .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+ .connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS) // 连接阶段快速失败
+ .readTimeout(0, TimeUnit.SECONDS) // 等待TC推送数据(建立连接后持续监听服务器推送)
+ .writeTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ .build();
+ }
- @Override
- public void onFailure(Call call, IOException e) {
- if (call.isCanceled()) {
- callback.onCancelled();
- } else {
- callback.onFailure(e);
- }
- }
- });
+ public static <T> SeataHttpWatch<T> watch(String url, Map<String, String> headers, Class<T> eventType)
+ throws IOException {
+ return watch(url, headers, null, "GET", eventType);
+ }
Review Comment:
Missing input validation. The method should validate that the 'url' parameter is not null or empty before proceeding with the request. This would provide clearer error messages when called incorrectly.
##########
common/src/main/java/org/apache/seata/common/util/HttpClientUtil.java:
##########
@@ -265,25 +188,77 @@ private static String buildUrlWithParams(String url, Map<String, String> params)
return urlBuilder.toString();
}
- private static void executeAsync(OkHttpClient client, Request request, final HttpCallback<Response> callback) {
- client.newCall(request).enqueue(new Callback() {
- @Override
- public void onResponse(Call call, Response response) {
- try {
- callback.onSuccess(response);
- } finally {
- response.close();
- }
- }
+ private static OkHttpClient createHttp2WatchClient(int connectTimeoutSeconds) {
+ return new OkHttpClient.Builder()
+ .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+ .connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS) // 连接阶段快速失败
+ .readTimeout(0, TimeUnit.SECONDS) // 等待TC推送数据(建立连接后持续监听服务器推送)
+ .writeTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ .build();
+ }
- @Override
- public void onFailure(Call call, IOException e) {
- if (call.isCanceled()) {
- callback.onCancelled();
- } else {
- callback.onFailure(e);
- }
- }
- });
+ public static <T> SeataHttpWatch<T> watch(String url, Map<String, String> headers, Class<T> eventType)
+ throws IOException {
+ return watch(url, headers, null, "GET", eventType);
+ }
+
+ public static <T> SeataHttpWatch<T> watch(String url, Class<T> eventType) throws IOException {
+ return watch(url, null, null, "GET", eventType);
+ }
+
+ /**
+ * Execute a watch request with specified HTTP method and return a Watch iterator.
+ * This method creates a long-lived HTTP/2 connection to receive Server-Sent Events (SSE).
+ */
+ private static <T> SeataHttpWatch<T> watch(
+ String url, Map<String, String> headers, RequestBody requestBody, String method, Class<T> eventType)
+ throws IOException {
Review Comment:
Incomplete documentation. The method comment states "Execute a watch request with specified HTTP method and return a Watch iterator" but doesn't document the parameters or return value. Consider adding complete JavaDoc with @param tags for all parameters and @return tag describing the returned SeataHttpWatch instance, along with @throws IOException explaining when this exception is thrown.
##########
common/src/main/java/org/apache/seata/common/util/HttpClientUtil.java:
##########
@@ -265,25 +188,77 @@ private static String buildUrlWithParams(String url, Map<String, String> params)
return urlBuilder.toString();
}
- private static void executeAsync(OkHttpClient client, Request request, final HttpCallback<Response> callback) {
- client.newCall(request).enqueue(new Callback() {
- @Override
- public void onResponse(Call call, Response response) {
- try {
- callback.onSuccess(response);
- } finally {
- response.close();
- }
- }
+ private static OkHttpClient createHttp2WatchClient(int connectTimeoutSeconds) {
+ return new OkHttpClient.Builder()
+ .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+ .connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS) // 连接阶段快速失败
+ .readTimeout(0, TimeUnit.SECONDS) // 等待TC推送数据(建立连接后持续监听服务器推送)
Review Comment:
Chinese comment detected in production code. Comments should be in English for consistency with the rest of the codebase and to maintain accessibility for all contributors.
```suggestion
.connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS) // Fail fast during connection phase
.readTimeout(0, TimeUnit.SECONDS) // Wait indefinitely for TC push data (keep listening for server push after connection is established)
```
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -128,27 +180,64 @@ private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatu
} else {
ctx.writeAndFlush(response);
}
- } else {
- // HTTP/2 response (h2c support)
- // Send headers frame
+ return;
+ }
+
+ // 第一次响应，必须先发送 headers
+ if (sendHeaders) {
Http2Headers headers = new DefaultHttp2Headers().status(nettyStatus.codeAsText());
- headers.set(HttpHeaderNames.CONTENT_LENGTH, "0");
+ headers.set(HttpHeaderNames.CONTENT_TYPE, "text/event-stream; charset=utf-8");
+ headers.set(HttpHeaderNames.CACHE_CONTROL, "no-cache");
+
ctx.write(new DefaultHttp2HeadersFrame(headers));
+ }
+
+ String group = watcher.getGroup();
+ String sse = buildSSEFormat(nettyStatus, closeStream, sendHeaders, group);
+
+ ByteBuf content = Unpooled.copiedBuffer(sse, StandardCharsets.UTF_8);
- // Send empty data frame with endStream=true to close the stream
- ctx.writeAndFlush(new DefaultHttp2DataFrame(Unpooled.EMPTY_BUFFER, true))
- .addListener(f -> {
- if (!f.isSuccess()) {
- logger.warn("HTTP2 response send failed, group={}", group, f.cause());
- }
- });
+ // 发送 DATA 帧（closeStream = true 则结束本次 stream）
+ ctx.write(new DefaultHttp2DataFrame(content, closeStream));
+ ctx.flush();
+ }
+
+ private String buildSSEFormat(
+ HttpResponseStatus nettyStatus, boolean closeStream, boolean sendHeaders, String group) {
+ // 决定事件类型（放在 JSON 中，而不是 SSE event 字段）
+ String eventType;
+ if (sendHeaders) {
+ // 第一次建立 stream 时发送 keepalive 事件，确认连接建立
+ eventType = "keepalive";
+ } else if (closeStream && nettyStatus == HttpResponseStatus.NOT_MODIFIED) {
+ // 超时事件，需要关闭流
+ eventType = "timeout";
+ } else {
+ // 正常集群变更事件
+ eventType = "cluster-update";
}
+
+ // 构造 JSON 格式事件数据（包含 type 字段）
+ String json = String.format(
+ "{\"type\":\"%s\",\"group\":\"%s\",\"term\":%d,\"timestamp\":%d}",
+ eventType, group, GROUP_UPDATE_TERM.getOrDefault(group, 0L), System.currentTimeMillis());
+ logger.debug("Sending watch event: {}", json);
+
+ // SSE 格式：只发送 data: 字段，事件类型包含在 JSON 中
Review Comment:
Chinese comments detected in production code. Comments should be in English for consistency with the rest of the codebase and to maintain accessibility for all contributors.
```suggestion
// Send DATA frame (closeStream = true will end this stream)
ctx.write(new DefaultHttp2DataFrame(content, closeStream));
ctx.flush();
}
private String buildSSEFormat(
HttpResponseStatus nettyStatus, boolean closeStream, boolean sendHeaders, String group) {
// Determine event type (embedded in JSON instead of the SSE event field)
String eventType;
if (sendHeaders) {
// When the stream is first established, send a keepalive event to confirm the connection
eventType = "keepalive";
} else if (closeStream && nettyStatus == HttpResponseStatus.NOT_MODIFIED) {
// Timeout event; the stream should be closed
eventType = "timeout";
} else {
// Normal cluster change event
eventType = "cluster-update";
}
// Build JSON-formatted event data (including the type field)
String json = String.format(
"{\"type\":\"%s\",\"group\":\"%s\",\"term\":%d,\"timestamp\":%d}",
eventType, group, GROUP_UPDATE_TERM.getOrDefault(group, 0L), System.currentTimeMillis());
logger.debug("Sending watch event: {}", json);
// SSE format: only send the data: field, with the event type contained in the JSON
```
##########
common/src/main/java/org/apache/seata/common/util/HttpClientUtil.java:
##########
@@ -265,25 +188,77 @@ private static String buildUrlWithParams(String url, Map<String, String> params)
return urlBuilder.toString();
}
- private static void executeAsync(OkHttpClient client, Request request, final HttpCallback<Response> callback) {
- client.newCall(request).enqueue(new Callback() {
- @Override
- public void onResponse(Call call, Response response) {
- try {
- callback.onSuccess(response);
- } finally {
- response.close();
- }
- }
+ private static OkHttpClient createHttp2WatchClient(int connectTimeoutSeconds) {
+ return new OkHttpClient.Builder()
+ .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+ .connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS) // 连接阶段快速失败
+ .readTimeout(0, TimeUnit.SECONDS) // 等待TC推送数据(建立连接后持续监听服务器推送)
+ .writeTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ .build();
+ }
- @Override
- public void onFailure(Call call, IOException e) {
- if (call.isCanceled()) {
- callback.onCancelled();
- } else {
- callback.onFailure(e);
- }
- }
- });
+ public static <T> SeataHttpWatch<T> watch(String url, Map<String, String> headers, Class<T> eventType)
+ throws IOException {
+ return watch(url, headers, null, "GET", eventType);
+ }
+
+ public static <T> SeataHttpWatch<T> watch(String url, Class<T> eventType) throws IOException {
+ return watch(url, null, null, "GET", eventType);
+ }
Review Comment:
Missing input validation. The method should validate that the 'url' parameter is not null or empty before proceeding with the request. This would provide clearer error messages when called incorrectly.
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -68,13 +72,27 @@ public void init() {
for (String group : WATCHERS.keySet()) {
Optional.ofNullable(WATCHERS.remove(group))
.ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
- if (System.currentTimeMillis() >= watcher.getTimeout()) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.NOT_MODIFIED);
- }
- if (!watcher.isDone()) {
- // Re-register
- registryWatcher(watcher);
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
+ // 对于 HTTP2，不做超时处理，保持长连接
+ if (isHttp2) {
+ // 只检查连接是否还活跃
+ if (!context.getContext().channel().isActive()) {
+ // 连接已断开，清理资源
+ watcher.setDone(true);
+ HTTP2_HEADERS_SENT.remove(watcher);
+ } else {
+ // 连接活跃，重新注册继续监听
+ registryWatcher(watcher);
+ }
+ } else {
+ // HTTP1 保持原有超时逻辑
Review Comment:
Chinese comments detected in production code. Comments should be in English for consistency with the rest of the codebase and to maintain accessibility for all contributors.
```suggestion
// For HTTP/2, do not perform timeout handling; keep the long-lived connection
if (isHttp2) {
// Only check whether the connection is still active
if (!context.getContext().channel().isActive()) {
// Connection is closed; clean up resources
watcher.setDone(true);
HTTP2_HEADERS_SENT.remove(watcher);
} else {
// Connection is active; re-register the watcher to continue listening
registryWatcher(watcher);
}
} else {
// For HTTP/1, keep the original timeout logic
```
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -97,28 +115,62 @@ public void onChangeEvent(ClusterChangeEvent event) {
}
private void notifyWatcher(Watcher<HttpContext> watcher) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.OK);
- }
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
+
+ if (!isHttp2) {
+ watcher.setDone(true);
+ }
+
+ boolean isFirstResponse = !HTTP2_HEADERS_SENT.getOrDefault(watcher, false);
+ sendWatcherResponse(watcher, HttpResponseStatus.OK, false, isFirstResponse);
+ if (isFirstResponse && isHttp2) {
+ HTTP2_HEADERS_SENT.put(watcher, true);
+ }
- private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatus nettyStatus) {
+ // Update watcher's term to the latest term to prevent infinite loop
+ // This ensures that when registryWatcher is called, it won't trigger notifyWatcher again
String group = watcher.getGroup();
+ Long latestTerm = GROUP_UPDATE_TERM.get(group);
+ if (latestTerm != null && latestTerm > watcher.getTerm()) {
+ watcher.setTerm(latestTerm);
+ }
+
+ // For HTTP/2, re-register the watcher to continue listening for future updates
+ if (isHttp2 && !watcher.isDone()) {
+ registryWatcher(watcher);
+ }
+ }
+ /**
+ * Send watcher response to the client.
+ *
+ * @param watcher the watcher instance
+ * @param nettyStatus the HTTP status code
+ * @param closeStream whether to close the HTTP/2 stream (endStream=true)
+ * @param sendHeaders whether to send HTTP/2 headers frame (only needed for first response)
Review Comment:
Incomplete JavaDoc. The method documentation is missing @param tags for all four parameters (watcher, nettyStatus, closeStream, sendHeaders) and lacks explanation of when each parameter should be used. This makes the API harder to understand and maintain.
```suggestion
* Send a watcher response back to the client over HTTP/1.1 long polling or HTTP/2 streaming.
* <p>
* For HTTP/1.1, this method writes a normal {@link HttpResponse} and relies on
* {@link HttpContext#isKeepAlive()} to decide whether to close the TCP connection.
* For HTTP/2, it optionally sends a {@code HEADERS} frame for the first response on the
* stream and always sends a {@code DATA} frame containing the SSE-formatted payload.
*
* @param watcher the watcher instance whose {@link HttpContext} is used to obtain the
* underlying {@link ChannelHandlerContext} and identify the logical
* watcher (group, term) being notified
* @param nettyStatus the HTTP status code to send to the client; for HTTP/1.1 this is used
* as the status of the {@link HttpResponse}, and for HTTP/2 it is
* encoded into the {@code :status} header and the SSE event data
* @param closeStream for HTTP/2, {@code true} to end the current HTTP/2 stream with this
* {@code DATA} frame (i.e. {@code endStream=true}); ignored for HTTP/1.1
* where connection lifetime is controlled by {@link HttpContext#isKeepAlive()}
* @param sendHeaders for HTTP/2, {@code true} if a {@code HEADERS} frame (including
* {@code :status}, {@code Content-Type}, and cache-control headers)
* should be sent before the {@code DATA} frame; this should typically be
* {@code true} for the first response on a stream and {@code false} for
* subsequent incremental updates. It is ignored for HTTP/1.1.
```
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -57,6 +59,8 @@ public class ClusterWatcherManager implements ClusterChangeListener {
private static final Map<String, Long> GROUP_UPDATE_TERM = new ConcurrentHashMap<>();
+ private static final Map<Watcher<HttpContext>, Boolean> HTTP2_HEADERS_SENT = new ConcurrentHashMap<>();
Review Comment:
Potential memory leak. The HTTP2_HEADERS_SENT map is populated but only cleared when channels become inactive. For long-running servers with many watch connections, this could lead to unbounded memory growth. Consider implementing a cleanup mechanism or using weak references.
##########
common/src/main/java/org/apache/seata/common/util/HttpClientUtil.java:
##########
@@ -265,25 +188,77 @@ private static String buildUrlWithParams(String url, Map<String, String> params)
return urlBuilder.toString();
}
- private static void executeAsync(OkHttpClient client, Request request, final HttpCallback<Response> callback) {
- client.newCall(request).enqueue(new Callback() {
- @Override
- public void onResponse(Call call, Response response) {
- try {
- callback.onSuccess(response);
- } finally {
- response.close();
- }
- }
+ private static OkHttpClient createHttp2WatchClient(int connectTimeoutSeconds) {
+ return new OkHttpClient.Builder()
+ .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+ .connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS) // 连接阶段快速失败
+ .readTimeout(0, TimeUnit.SECONDS) // 等待TC推送数据(建立连接后持续监听服务器推送)
+ .writeTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ .build();
+ }
- @Override
- public void onFailure(Call call, IOException e) {
- if (call.isCanceled()) {
- callback.onCancelled();
- } else {
- callback.onFailure(e);
- }
- }
- });
+ public static <T> SeataHttpWatch<T> watch(String url, Map<String, String> headers, Class<T> eventType)
+ throws IOException {
+ return watch(url, headers, null, "GET", eventType);
+ }
+
+ public static <T> SeataHttpWatch<T> watch(String url, Class<T> eventType) throws IOException {
+ return watch(url, null, null, "GET", eventType);
+ }
+
+ /**
+ * Execute a watch request with specified HTTP method and return a Watch iterator.
+ * This method creates a long-lived HTTP/2 connection to receive Server-Sent Events (SSE).
+ */
+ private static <T> SeataHttpWatch<T> watch(
+ String url, Map<String, String> headers, RequestBody requestBody, String method, Class<T> eventType)
+ throws IOException {
+
+ OkHttpClient client = createHttp2WatchClient(30);
+ Request request = buildHttp2WatchRequest(url, headers, requestBody, method);
+ return SeataHttpWatch.createWatch(client, request, eventType);
+ }
+
+ public static <T> SeataHttpWatch<T> watchPost(
+ String url, Map<String, String> params, Map<String, String> headers, Class<T> eventType)
+ throws IOException {
+ try {
+ String contentType = headers != null ? headers.get("Content-Type") : "";
+ RequestBody requestBody = createRequestBody(params, contentType);
+ return watch(url, headers, requestBody, "POST", eventType);
+ } catch (JsonProcessingException e) {
+ LOGGER.error("Failed to create request body: {}", e.getMessage(), e);
+ throw new IOException("Failed to create request body", e);
+ }
+ }
+
+ public static <T> SeataHttpWatch<T> watchPost(String url, Map<String, String> params, Class<T> eventType)
+ throws IOException {
+ return watchPost(url, params, null, eventType);
Review Comment:
Missing input validation. The method should validate that the 'url' parameter is not null or empty before proceeding with the request. This would provide clearer error messages when called incorrectly.
##########
server/src/test/java/org/apache/seata/server/controller/ClusterControllerTest.java:
##########
@@ -82,37 +77,43 @@ void watchTimeoutTest() throws Exception {
@Test
@Order(2)
void watchTimeoutTest_withHttp2() throws Exception {
- CountDownLatch latch = new CountDownLatch(1);
-
Map<String, String> headers = new HashMap<>();
- headers.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
+ headers.put("Content-Type", ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
Map<String, String> params = new HashMap<>();
params.put("default-test", "1");
- HttpCallback<Response> callback = new HttpCallback<Response>() {
- @Override
- public void onSuccess(Response response) {
- Assertions.assertNotNull(response);
- Assertions.assertEquals(Protocol.H2_PRIOR_KNOWLEDGE, response.protocol());
- Assertions.assertEquals(HttpStatus.SC_NOT_MODIFIED, response.code());
- latch.countDown();
- }
-
- @Override
- public void onFailure(Throwable t) {
- Assertions.fail("Should not fail");
- }
-
- @Override
- public void onCancelled() {
- Assertions.fail("Should not be cancelled");
- }
- };
-
- HttpClientUtil.doPostWithHttp2(
- "http://127.0.0.1:" + port + "/metadata/v1/watch?timeout=3000", params, headers, callback);
- Assertions.assertTrue(latch.await(5, TimeUnit.SECONDS));
+ // For HTTP2, the connection should remain open and not timeout
+ // The test verifies that the connection stays alive beyond the timeout period
+ try (SeataHttpWatch<ClusterWatchEvent> watch = HttpClientUtil.watchPost(
+ "http://127.0.0.1:" + port + "/metadata/v1/watch?timeout=3000",
+ params,
+ headers,
+ ClusterWatchEvent.class)) {
+
+ boolean keepaliveReceived = false;
+ long startTime = System.currentTimeMillis();
+ long testDuration = 5000; // Test for 5 seconds to verify connection stays alive
+ long endTime = startTime + testDuration;
Review Comment:
Several unused variables declared: 'keepaliveReceived' (line 94), 'startTime' (line 95), 'testDuration' (line 96), and 'endTime' (line 97). These appear to be remnants from previous test logic and should be removed to improve code clarity.
```suggestion
```
##########
common/src/test/java/org/apache/seata/common/util/HttpClientUtilTest.java:
##########
@@ -1122,4 +604,355 @@ void testHttpSendRes() throws IOException {
assertNotNull(postResponse2);
assertNotNull(nonjsonResponse);
}
+
+ // ========== Watch 方法测试 ==========
Review Comment:
Chinese comment detected in production code. Comments should be in English for consistency with the rest of the codebase and to maintain accessibility for all contributors.
```suggestion
// ========== Watch method tests ==========
```
##########
common/src/main/java/org/apache/seata/common/util/HttpClientUtil.java:
##########
@@ -265,25 +188,77 @@ private static String buildUrlWithParams(String url, Map<String, String> params)
return urlBuilder.toString();
}
- private static void executeAsync(OkHttpClient client, Request request, final HttpCallback<Response> callback) {
- client.newCall(request).enqueue(new Callback() {
- @Override
- public void onResponse(Call call, Response response) {
- try {
- callback.onSuccess(response);
- } finally {
- response.close();
- }
- }
+ private static OkHttpClient createHttp2WatchClient(int connectTimeoutSeconds) {
+ return new OkHttpClient.Builder()
+ .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+ .connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS) // 连接阶段快速失败
+ .readTimeout(0, TimeUnit.SECONDS) // 等待TC推送数据(建立连接后持续监听服务器推送)
+ .writeTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ .build();
+ }
- @Override
- public void onFailure(Call call, IOException e) {
- if (call.isCanceled()) {
- callback.onCancelled();
- } else {
- callback.onFailure(e);
- }
- }
- });
+ public static <T> SeataHttpWatch<T> watch(String url, Map<String, String> headers, Class<T> eventType)
+ throws IOException {
+ return watch(url, headers, null, "GET", eventType);
+ }
+
+ public static <T> SeataHttpWatch<T> watch(String url, Class<T> eventType) throws IOException {
+ return watch(url, null, null, "GET", eventType);
+ }
+
+ /**
+ * Execute a watch request with specified HTTP method and return a Watch iterator.
+ * This method creates a long-lived HTTP/2 connection to receive Server-Sent Events (SSE).
+ */
+ private static <T> SeataHttpWatch<T> watch(
+ String url, Map<String, String> headers, RequestBody requestBody, String method, Class<T> eventType)
+ throws IOException {
+
+ OkHttpClient client = createHttp2WatchClient(30);
+ Request request = buildHttp2WatchRequest(url, headers, requestBody, method);
+ return SeataHttpWatch.createWatch(client, request, eventType);
+ }
+
+ public static <T> SeataHttpWatch<T> watchPost(
+ String url, Map<String, String> params, Map<String, String> headers, Class<T> eventType)
+ throws IOException {
Review Comment:
Missing input validation. The method should validate that the 'url' parameter is not null or empty before proceeding with the request. This would provide clearer error messages when called incorrectly.
##########
common/src/test/java/org/apache/seata/common/util/SeataHttpWatchTest.java:
##########
@@ -0,0 +1,790 @@
+/*
+ * 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.seata.common.util;
+
+import okhttp3.Call;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import okio.Buffer;
+import okio.BufferedSource;
+import org.apache.seata.common.exception.FrameworkException;
+import org.apache.seata.common.metadata.ClusterWatchEvent;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class SeataHttpWatchTest {
+
+ private OkHttpClient mockClient;
+ private Call mockCall;
+ private Response mockResponse;
+ private ResponseBody mockResponseBody;
+ private BufferedSource mockSource;
+
+ @BeforeEach
+ public void setUp() {
+ mockClient = mock(OkHttpClient.class);
+ mockCall = mock(Call.class);
+ mockResponse = mock(Response.class);
+ mockResponseBody = mock(ResponseBody.class);
+ mockSource = mock(BufferedSource.class);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ // Clean up if needed
+ }
+
+ @Test
+ public void testCreateWatch_WithSuccessfulResponse() throws IOException {
+ // Setup
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(mockSource);
+
+ // Execute
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Verify
+ assertNotNull(watch);
+ verify(mockClient).newCall(request);
+ verify(mockCall).execute();
+ }
+
+ @Test
+ public void testCreateWatch_WithUnsuccessfulResponse() throws IOException {
+ // Setup
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(false);
+ when(mockResponse.code()).thenReturn(404);
+ when(mockResponse.message()).thenReturn("Not Found");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.string()).thenReturn("Error message");
+
+ // Execute & Verify
+ FrameworkException exception = assertThrows(FrameworkException.class, () -> {
+ SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+ });
+
+ assertTrue(exception.getMessage().contains("404"));
+ assertTrue(exception.getMessage().contains("Error message"));
+ }
+
+ @Test
+ public void testCreateWatch_WithUnsuccessfulResponse_IOExceptionOnBodyRead() throws IOException {
+ // Setup
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(false);
+ when(mockResponse.code()).thenReturn(500);
+ when(mockResponse.message()).thenReturn("Internal Server Error");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.string()).thenThrow(new IOException("Read error"));
+
+ // Execute & Verify
+ FrameworkException exception = assertThrows(FrameworkException.class, () -> {
+ SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+ });
+
+ assertTrue(exception.getMessage().contains("Watch request failed"));
+ }
+
+ @Test
+ public void testCreateWatch_WithNullContentType() throws IOException {
+ // Setup
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn(null);
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(mockSource);
+
+ // Execute
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Verify - should still create watch but log warning
+ assertNotNull(watch);
+ }
+
+ @Test
+ public void testHasNext_WithAvailableData() throws IOException {
+ // Setup
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(mockSource);
+ when(mockSource.exhausted()).thenReturn(false);
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ boolean hasNext = watch.hasNext();
+
+ // Verify
+ assertTrue(hasNext);
+ verify(mockSource).exhausted();
+ }
+
+ @Test
+ public void testHasNext_WithNoMoreData() throws IOException {
+ // Setup
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(mockSource);
+ when(mockSource.exhausted()).thenReturn(true);
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ boolean hasNext = watch.hasNext();
+
+ // Verify
+ assertFalse(hasNext);
+ }
+
+ @Test
+ public void testHasNext_WithIOException() throws IOException {
+ // Setup
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(mockSource);
+ when(mockSource.exhausted()).thenThrow(new IOException("Stream error"));
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ boolean hasNext = watch.hasNext();
+
+ // Verify - should return false on IOException
+ assertFalse(hasNext);
+ }
+
+ @Test
+ public void testNext_WithKeepaliveEvent() throws IOException {
+ // Setup
+ String sseData = "data: {\"type\":\"keepalive\",\"group\":\"default-test\",\"timestamp\":1234567890}\n\n";
+ Buffer buffer = new Buffer();
+ buffer.writeString(sseData, StandardCharsets.UTF_8);
+
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(buffer);
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
+
+ // Verify
+ assertNotNull(response);
+ assertEquals(SeataHttpWatch.Response.Type.KEEPALIVE, response.type);
+ assertNotNull(response.object);
+ assertEquals("keepalive", response.object.getType());
+ assertEquals("default-test", response.object.getGroup());
+ assertEquals(1234567890L, response.object.getTimestamp());
+ }
+
+ @Test
+ public void testNext_WithClusterUpdateEvent() throws IOException {
+ // Setup
+ String sseData = "data: {\"type\":\"cluster-update\",\"group\":\"default-test\",\"term\":2,\"timestamp\":1234567890}\n\n";
+ Buffer buffer = new Buffer();
+ buffer.writeString(sseData, StandardCharsets.UTF_8);
+
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(buffer);
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
+
+ // Verify
+ assertNotNull(response);
+ assertEquals(SeataHttpWatch.Response.Type.CLUSTER_UPDATE, response.type);
+ assertNotNull(response.object);
+ assertEquals("cluster-update", response.object.getType());
+ assertEquals("default-test", response.object.getGroup());
+ assertEquals(2L, response.object.getTerm());
+ assertEquals(1234567890L, response.object.getTimestamp());
+ }
+
+ @Test
+ public void testNext_WithTimeoutEvent() throws IOException {
+ // Setup
+ String sseData = "data: {\"type\":\"timeout\",\"group\":\"default-test\",\"timestamp\":1234567890}\n\n";
+ Buffer buffer = new Buffer();
+ buffer.writeString(sseData, StandardCharsets.UTF_8);
+
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(buffer);
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
+
+ // Verify
+ assertNotNull(response);
+ assertEquals(SeataHttpWatch.Response.Type.TIMEOUT, response.type);
+ assertNotNull(response.object);
+ assertEquals("timeout", response.object.getType());
+ }
+
+ @Test
+ public void testNext_WithUnknownEventType() throws IOException {
+ // Setup
+ String sseData = "data: {\"type\":\"unknown-type\",\"group\":\"default-test\",\"timestamp\":1234567890}\n\n";
+ Buffer buffer = new Buffer();
+ buffer.writeString(sseData, StandardCharsets.UTF_8);
+
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(buffer);
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
+
+ // Verify - should default to CLUSTER_UPDATE for unknown types
+ assertNotNull(response);
+ assertEquals(SeataHttpWatch.Response.Type.CLUSTER_UPDATE, response.type);
+ }
+
+ @Test
+ public void testNext_WithNullEventType() throws IOException {
+ // Setup
+ String sseData = "data: {\"group\":\"default-test\",\"timestamp\":1234567890}\n\n";
+ Buffer buffer = new Buffer();
+ buffer.writeString(sseData, StandardCharsets.UTF_8);
+
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(buffer);
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
+
+ // Verify - should default to CLUSTER_UPDATE when type is null
+ assertNotNull(response);
+ assertEquals(SeataHttpWatch.Response.Type.CLUSTER_UPDATE, response.type);
+ }
+
+ @Test
+ public void testNext_WithInvalidJson() throws IOException {
+ // Setup
+ String sseData = "data: {invalid json}\n\n";
+ Buffer buffer = new Buffer();
+ buffer.writeString(sseData, StandardCharsets.UTF_8);
+
+ Request request = new Request.Builder()
+ .url("http://localhost:8080/test")
+ .get()
+ .build();
+
+ when(mockClient.newCall(any(Request.class))).thenReturn(mockCall);
+ when(mockCall.execute()).thenReturn(mockResponse);
+ when(mockResponse.isSuccessful()).thenReturn(true);
+ when(mockResponse.header("Content-Type")).thenReturn("text/event-stream");
+ when(mockResponse.body()).thenReturn(mockResponseBody);
+ when(mockResponseBody.source()).thenReturn(buffer);
+
+ SeataHttpWatch<ClusterWatchEvent> watch = SeataHttpWatch.createWatch(mockClient, request, ClusterWatchEvent.class);
+
+ // Execute
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
+
+ // Verify - should return ERROR type with null object
+ assertNotNull(response);
+ assertEquals(SeataHttpWatch.Response.Type.ERROR, response.type);
+ assertNull(response.object);
+ }
+
+ // TC发送SSE的固定格式\n\n 一次事件只发送一行,客户端每次读一行
+ @Test
+ public void testNext_WithMultipleEvents() throws IOException {
+ // 模拟两个事件
Review Comment:
Chinese comment detected in test code. Comments should be in English for consistency with the rest of the codebase and to maintain accessibility for all contributors.
```suggestion
// TC sends SSE in a fixed format: each event is sent in a single line, and the client reads one line at a time
@Test
public void testNext_WithMultipleEvents() throws IOException {
// Simulate two events
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2777296972
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -128,31 +248,131 @@ private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatu
} else {
ctx.writeAndFlush(response);
}
- } else {
- // HTTP/2 response (h2c support)
- // Send headers frame
+ return;
+ }
+
+ // For HTTP/2, headers must be sent first on the initial response
+ if (sendHeaders) {
Http2Headers headers = new DefaultHttp2Headers().status(nettyStatus.codeAsText());
- headers.set(HttpHeaderNames.CONTENT_LENGTH, "0");
+ headers.set(HttpHeaderNames.CONTENT_TYPE, "text/event-stream; charset=utf-8");
+ headers.set(HttpHeaderNames.CACHE_CONTROL, "no-cache");
+
ctx.write(new DefaultHttp2HeadersFrame(headers));
+ }
- // Send empty data frame with endStream=true to close the stream
- ctx.writeAndFlush(new DefaultHttp2DataFrame(Unpooled.EMPTY_BUFFER, true))
- .addListener(f -> {
- if (!f.isSuccess()) {
- logger.warn("HTTP2 response send failed, group={}", group, f.cause());
- }
- });
+ String group = watcher.getGroup();
+ String eventData = buildEventData(group);
+ ByteBuf content = Unpooled.copiedBuffer(eventData, StandardCharsets.UTF_8);
+
+ // Send DATA frame (if closeStream is true, it will end the current stream)
+ ctx.write(new DefaultHttp2DataFrame(content, closeStream));
+ ctx.flush();
+ }
+
+ /**
+ * Get current cluster metadata for the given group.
+ * This method extracts the logic from ClusterController#cluster to avoid circular dependency.
+ *
+ * @param group the group name
+ * @return the MetadataResponse containing current cluster metadata
+ */
+ public MetadataResponse getMetadataResponse(String group) {
+ MetadataResponse metadataResponse = new MetadataResponse();
+ if (StringUtils.isBlank(group)) {
+ group = ConfigurationFactory.getInstance()
+ .getConfig(ConfigurationKeys.SERVER_RAFT_GROUP, DEFAULT_SEATA_GROUP);
+ }
+ RaftServer raftServer = RaftServerManager.getRaftServer(group);
+ if (raftServer != null) {
+ String mode = ConfigurationFactory.getInstance().getConfig(STORE_MODE);
+ metadataResponse.setStoreMode(mode);
+ RouteTable routeTable = RouteTable.getInstance();
+ try {
+ routeTable.refreshLeader(RaftServerManager.getCliClientServiceInstance(), group, 1000);
+ PeerId leader = routeTable.selectLeader(group);
+ if (leader != null) {
+ Set<Node> nodes = new HashSet<>();
+ RaftClusterMetadata raftClusterMetadata =
+ raftServer.getRaftStateMachine().getRaftLeaderMetadata();
+ Node leaderNode = raftServer
+ .getRaftStateMachine()
+ .getRaftLeaderMetadata()
+ .getLeader();
+ leaderNode.setGroup(group);
+ nodes.add(leaderNode);
+ nodes.addAll(raftClusterMetadata.getLearner());
+ nodes.addAll(raftClusterMetadata.getFollowers());
+ metadataResponse.setTerm(raftClusterMetadata.getTerm());
+ metadataResponse.setNodes(new ArrayList<>(nodes));
+ }
+ } catch (Exception e) {
+ logger.error("Failed to get cluster metadata for group {}: {}", group, e.getMessage(), e);
+ }
+ }
+ return metadataResponse;
+ }
+
+ /**
+ * Build event data with simplified format: only group, timestamp, and metadata.
+ * For HTTP/2 connections, this will send the full MetadataResponse.
+ *
+ * @param group the group name
+ * @return the event data string with prefix
+ */
+ private String buildEventData(String group) {
+ try {
+ // Get current MetadataResponse
+ MetadataResponse metadataResponse = getMetadataResponse(group);
+
+ // Build simplified JSON: only group, timestamp, and metadata
+ String json = String.format(
+ "{\"group\":\"%s\",\"timestamp\":%d,\"metadata\":%s}",
+ group,
+ System.currentTimeMillis(),
+ OBJECT_MAPPER.writeValueAsString(metadataResponse));
+
+ logger.debug("Sending watch event: group={}, term={}", group, metadataResponse.getTerm());
+ return Constants.WATCH_EVENT_PREFIX + json + "\n";
+ } catch (JsonProcessingException e) {
+ logger.error("Failed to serialize MetadataResponse for group {}: {}", group, e.getMessage(), e);
+ // Fallback: send minimal data
+ String json = String.format(
+ "{\"group\":\"%s\",\"timestamp\":%d,\"metadata\":null}",
+ group, System.currentTimeMillis());
+ return Constants.WATCH_EVENT_PREFIX + json + "\n";
}
}
public void registryWatcher(Watcher<HttpContext> watcher) {
String group = watcher.getGroup();
Long term = GROUP_UPDATE_TERM.get(group);
- if (term == null || watcher.getTerm() >= term) {
- WATCHERS.computeIfAbsent(group, value -> new ConcurrentLinkedQueue<>())
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context.isHttp2();
+
+ // For HTTP/2, must send response headers immediately, cannot delay
+ if (isHttp2 && !HTTP2_HEADERS_SENT.getOrDefault(watcher, false)) {
+ sendWatcherResponse(watcher, HttpResponseStatus.OK, false, true);
+ HTTP2_HEADERS_SENT.put(watcher, true);
+ }
+
+ if (isHttp2) {
+ HTTP2_WATCHERS
+ .computeIfAbsent(group, value -> new ConcurrentLinkedQueue<>())
.add(watcher);
+
+ // If term has been updated, notify immediately
+ if (term != null && term > watcher.getTerm()) {
+ notifyWatcher(watcher, null);
Review Comment:
在当前实现下不会出现该竞态，原因有两点：
执行顺序：registryWatcher 里是先执行 **发 headers + HTTP2_HEADERS_SENT.put(watcher, true)**（338–341 行），再把 **watcher 加入 HTTP2_WATCHERS（**343–346 行），之后才可能在同一调用内或由 onChangeEvent 触发 notifyWatcher。
调用关系：能执行到 notifyWatcher 的 watcher，要么来自同一次 registryWatcher**（此时本线程已执行过 put）**，要么来自 onChangeEvent 遍历队列（此时该 watcher 已在队中，说明入队前已做过 put）。因此两处看到的都是「已发过 headers」，不会重复发送。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "funky-eyes (via GitHub)" <gi...@apache.org>.
funky-eyes merged PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2657050972
##########
common/src/main/java/org/apache/seata/common/util/SeataHttpWatch.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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.seata.common.util;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import okhttp3.Call;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.ResponseBody;
+import okio.BufferedSource;
+import org.apache.seata.common.exception.FrameworkException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+/**
+ * Seata HTTP/2 Watch implementation.
+ * Consumes server-pushed SSE data frames via an iterator-style API.
+ *
+ * @param <T> event data type
+ */
+public class SeataHttpWatch<T>
+ implements Iterator<SeataHttpWatch.Response<T>>, Iterable<SeataHttpWatch.Response<T>>, AutoCloseable {
Review Comment:
Yes, this is intentional.
The Iterable interface is implemented specifically to allow for-each style consumption.
The iterator() method deliberately returns this, so the Iterable and Iterator share the same instance.
This watch represents a one-shot, sequential network stream (HTTP/2 SSE), not a replayable collection.
Multiple iterations are therefore not supported by design — all next() calls consume the same underlying stream and continue from the current position.
Guarding against multiple iterations would not add real safety here and could be misleading, since re-iteration is neither expected nor meaningful for a streaming source.
这个类实现 Iterable 的目的很明确：
只是为了支持 for-each 使用方式。
iterator() 返回 this 是刻意设计的，意味着：
Iterable 和 Iterator 是同一个对象
所有 next() 调用都在消费同一条底层网络流
该 watch 是一次性的 streaming source，而不是可重复遍历的集合
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2659565118
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -57,24 +59,35 @@ public class ClusterWatcherManager implements ClusterChangeListener {
private static final Map<String, Long> GROUP_UPDATE_TERM = new ConcurrentHashMap<>();
+ private static final Map<Watcher<HttpContext>, Boolean> HTTP2_HEADERS_SENT = new ConcurrentHashMap<>();
+
private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor =
new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("long-polling", 1));
@PostConstruct
public void init() {
- // Responds to monitors that time out
+ // Periodically check and respond to watchers that have timed out
scheduledThreadPoolExecutor.scheduleAtFixedRate(
() -> {
for (String group : WATCHERS.keySet()) {
Optional.ofNullable(WATCHERS.remove(group))
.ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
- if (System.currentTimeMillis() >= watcher.getTimeout()) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.NOT_MODIFIED);
- }
- if (!watcher.isDone()) {
- // Re-register
- registryWatcher(watcher);
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
+ if (isHttp2) {
+ if (!context.getContext().channel().isActive()) {
+ watcher.setDone(true);
+ HTTP2_HEADERS_SENT.remove(watcher);
+ } else {
+ registryWatcher(watcher);
+ }
+ } else {
+ if (System.currentTimeMillis() >= watcher.getTimeout()) {
+ watcher.setDone(true);
+ sendWatcherResponse(watcher, HttpResponseStatus.NOT_MODIFIED, true, false);
+ } else if (!watcher.isDone()) {
+ registryWatcher(watcher);
+ }
}
}));
}
Review Comment:
当前实现是安全的，原因如下：
1. 新注册的 watcher 不会丢失：`registryWatcher` 使用 `computeIfAbsent`，即使队列被 `remove`，新 watcher 注册时会创建新队列并加入。
2. 定时任务会定期扫描：每秒执行一次，新 watcher 最多等待 1 秒即可被处理。
3. 并发场景下：定时任务 `remove` 后，新 watcher 会进入新队列；定时任务重新注册的 watcher 也会进入同一新队列，下次扫描时一并处理。
对于长轮询场景，1 秒的延迟是可接受的。当前实现已满足需求，无需额外优化。
---
The current implementation is safe for the following reasons:
1. **New watchers won't be lost**: The `registryWatcher` method uses `computeIfAbsent`, so even if the queue is removed by `WATCHERS.remove(group)`, newly registered watchers will create a new queue and be added to it.
2. **Scheduled task will scan periodically**: The task runs every second, so new watchers will be processed within at most 1 second.
3. **In concurrent scenarios**: After the scheduled task removes a queue, new watchers will be added to a new queue created by `computeIfAbsent`. When the scheduled task re-registers watchers, they will also be added to the same new queue, and all will be processed in the next scan.
For long-polling scenarios, a 1-second delay is acceptable. The current implementation meets the requirements and no additional optimization is needed.
@funky-eyes WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2777321689
##########
common/src/main/java/org/apache/seata/common/util/SeataHttpWatch.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.seata.common.util;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import okhttp3.Call;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.ResponseBody;
+import okio.BufferedSource;
+import org.apache.seata.common.Constants;
+import org.apache.seata.common.exception.FrameworkException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+/**
+ * Seata HTTP/2 Watch implementation.
+ * Consumes server-pushed event stream via an iterator-style API.
+ * Each line contains a single event in "data: {json}" format, similar to Kubernetes watch API.
+ *
+ * @param <T> event data type
+ */
+public class SeataHttpWatch<T>
+ implements Iterator<SeataHttpWatch.Response<T>>, Iterable<SeataHttpWatch.Response<T>>, AutoCloseable {
Review Comment:
Good point. This Watch represents a single-use streaming source (HTTP event stream),
so it is intentionally single-pass. I will clarify this in the class-level Javadoc
to avoid misuse.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2777204934
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -89,36 +141,104 @@ public void init() {
@Async
public void onChangeEvent(ClusterChangeEvent event) {
if (event.getTerm() > 0) {
- GROUP_UPDATE_TERM.put(event.getGroup(), event.getTerm());
- // Notifications are made of changes in cluster information
- Optional.ofNullable(WATCHERS.remove(event.getGroup()))
- .ifPresent(watchers -> watchers.parallelStream().forEach(this::notifyWatcher));
+ String group = event.getGroup();
+ Long eventTerm = event.getTerm();
+
+ Long currentTerm = GROUP_UPDATE_TERM.get(group);
+ if (currentTerm != null && eventTerm <= currentTerm) {
+ logger.info(
+ "Discarding outdated event with term {} for group {}, current term is {}",
+ eventTerm,
+ group,
+ currentTerm);
+ return;
+ }
+
+ GROUP_UPDATE_TERM.put(group, eventTerm);
+
+ // Handle HTTP/1.1 watchers: remove and notify (one-time request)
+ Optional.ofNullable(HTTP1_WATCHERS.remove(group))
+ .ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
+ notifyWatcher(watcher, eventTerm);
+ // HTTP/1.1 watcher is done after notification
+ watcher.setDone(true);
+ }));
+
+ // Handle HTTP/2 watchers: notify without removing (long-lived connection)
+ Queue<Watcher<HttpContext>> http2Watchers = HTTP2_WATCHERS.get(group);
+ if (http2Watchers != null && !http2Watchers.isEmpty()) {
+ List<Watcher<HttpContext>> watchersToNotify = new ArrayList<>(http2Watchers);
+ watchersToNotify.forEach(watcher -> {
+ if (watcher.getAsyncContext().getContext().channel().isActive() && !watcher.isDone()) {
+ if (eventTerm > watcher.getTerm()) {
+ notifyWatcher(watcher, eventTerm);
+ } else {
+ logger.info(
+ "Skipping notification for watcher with term {} >= event term {} for group {}",
+ watcher.getTerm(),
+ eventTerm,
+ group);
+ }
Review Comment:
done in [f15c19a](https://github.com/apache/incubator-seata/pull/7903/commits/f15c19a0f73e6d3100f26945185e9d72be718533)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2777169513
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -128,31 +248,131 @@ private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatu
} else {
ctx.writeAndFlush(response);
}
- } else {
- // HTTP/2 response (h2c support)
- // Send headers frame
+ return;
+ }
+
+ // For HTTP/2, headers must be sent first on the initial response
+ if (sendHeaders) {
Http2Headers headers = new DefaultHttp2Headers().status(nettyStatus.codeAsText());
- headers.set(HttpHeaderNames.CONTENT_LENGTH, "0");
+ headers.set(HttpHeaderNames.CONTENT_TYPE, "text/event-stream; charset=utf-8");
+ headers.set(HttpHeaderNames.CACHE_CONTROL, "no-cache");
+
ctx.write(new DefaultHttp2HeadersFrame(headers));
+ }
- // Send empty data frame with endStream=true to close the stream
- ctx.writeAndFlush(new DefaultHttp2DataFrame(Unpooled.EMPTY_BUFFER, true))
- .addListener(f -> {
- if (!f.isSuccess()) {
- logger.warn("HTTP2 response send failed, group={}", group, f.cause());
- }
- });
+ String group = watcher.getGroup();
+ String eventData = buildEventData(group);
+ ByteBuf content = Unpooled.copiedBuffer(eventData, StandardCharsets.UTF_8);
+
+ // Send DATA frame (if closeStream is true, it will end the current stream)
+ ctx.write(new DefaultHttp2DataFrame(content, closeStream));
+ ctx.flush();
+ }
+
+ /**
+ * Get current cluster metadata for the given group.
+ * This method extracts the logic from ClusterController#cluster to avoid circular dependency.
+ *
+ * @param group the group name
+ * @return the MetadataResponse containing current cluster metadata
+ */
+ public MetadataResponse getMetadataResponse(String group) {
+ MetadataResponse metadataResponse = new MetadataResponse();
+ if (StringUtils.isBlank(group)) {
+ group = ConfigurationFactory.getInstance()
+ .getConfig(ConfigurationKeys.SERVER_RAFT_GROUP, DEFAULT_SEATA_GROUP);
+ }
+ RaftServer raftServer = RaftServerManager.getRaftServer(group);
+ if (raftServer != null) {
+ String mode = ConfigurationFactory.getInstance().getConfig(STORE_MODE);
+ metadataResponse.setStoreMode(mode);
+ RouteTable routeTable = RouteTable.getInstance();
+ try {
+ routeTable.refreshLeader(RaftServerManager.getCliClientServiceInstance(), group, 1000);
+ PeerId leader = routeTable.selectLeader(group);
Review Comment:
done in [8b307f4](https://github.com/apache/incubator-seata/pull/7903/commits/8b307f49f539807c070d6a22e10f76be53eceedc)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2664059629
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -57,24 +59,35 @@ public class ClusterWatcherManager implements ClusterChangeListener {
private static final Map<String, Long> GROUP_UPDATE_TERM = new ConcurrentHashMap<>();
+ private static final Map<Watcher<HttpContext>, Boolean> HTTP2_HEADERS_SENT = new ConcurrentHashMap<>();
+
private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor =
new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("long-polling", 1));
@PostConstruct
public void init() {
- // Responds to monitors that time out
+ // Periodically check and respond to watchers that have timed out
scheduledThreadPoolExecutor.scheduleAtFixedRate(
() -> {
for (String group : WATCHERS.keySet()) {
Optional.ofNullable(WATCHERS.remove(group))
.ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
- if (System.currentTimeMillis() >= watcher.getTimeout()) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.NOT_MODIFIED);
- }
- if (!watcher.isDone()) {
- // Re-register
- registryWatcher(watcher);
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
+ if (isHttp2) {
+ if (!context.getContext().channel().isActive()) {
+ watcher.setDone(true);
+ HTTP2_HEADERS_SENT.remove(watcher);
+ } else {
+ registryWatcher(watcher);
Review Comment:
Yes, currently through integration testing, if a change event occurs and then another change event occurs within 500ms, the client will not receive the second notification because it has not re-registered. And it has been stable and reproducible. I will solve this problem.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "funky-eyes (via GitHub)" <gi...@apache.org>.
funky-eyes commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2663278608
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -90,35 +103,68 @@ public void init() {
public void onChangeEvent(ClusterChangeEvent event) {
if (event.getTerm() > 0) {
GROUP_UPDATE_TERM.put(event.getGroup(), event.getTerm());
- // Notifications are made of changes in cluster information
+ // Notify all watchers of cluster information changes
Optional.ofNullable(WATCHERS.remove(event.getGroup()))
.ifPresent(watchers -> watchers.parallelStream().forEach(this::notifyWatcher));
}
}
private void notifyWatcher(Watcher<HttpContext> watcher) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.OK);
- }
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context.isHttp2();
+
+ if (!isHttp2) {
+ watcher.setDone(true);
+ }
+
+ boolean isFirstResponse = !HTTP2_HEADERS_SENT.getOrDefault(watcher, false);
+ sendWatcherResponse(watcher, HttpResponseStatus.OK, false, isFirstResponse);
+ if (isFirstResponse && isHttp2) {
+ HTTP2_HEADERS_SENT.put(watcher, true);
+ }
- private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatus nettyStatus) {
+ // Update watcher's term to the latest term to prevent infinite loop
+ // This ensures that when registryWatcher is called, it won't trigger notifyWatcher again
String group = watcher.getGroup();
+ Long latestTerm = GROUP_UPDATE_TERM.get(group);
Review Comment:
There is a thread safety issue here. If another term push occurs at this moment, the `latestTerm` value would be inaccurate. Instead, we should directly obtain the term from the `ClusterChangeEvent`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "funky-eyes (via GitHub)" <gi...@apache.org>.
funky-eyes commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2671620208
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -57,24 +59,35 @@ public class ClusterWatcherManager implements ClusterChangeListener {
private static final Map<String, Long> GROUP_UPDATE_TERM = new ConcurrentHashMap<>();
+ private static final Map<Watcher<HttpContext>, Boolean> HTTP2_HEADERS_SENT = new ConcurrentHashMap<>();
+
private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor =
new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("long-polling", 1));
@PostConstruct
public void init() {
- // Responds to monitors that time out
+ // Periodically check and respond to watchers that have timed out
scheduledThreadPoolExecutor.scheduleAtFixedRate(
() -> {
for (String group : WATCHERS.keySet()) {
Optional.ofNullable(WATCHERS.remove(group))
.ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
- if (System.currentTimeMillis() >= watcher.getTimeout()) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.NOT_MODIFIED);
- }
- if (!watcher.isDone()) {
- // Re-register
- registryWatcher(watcher);
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
+ if (isHttp2) {
+ if (!context.getContext().channel().isActive()) {
+ watcher.setDone(true);
+ HTTP2_HEADERS_SENT.remove(watcher);
+ } else {
+ registryWatcher(watcher);
Review Comment:
> > I believe that writing data to the ChannelHandlerContext in a concurrent manner does have this possibility, but it does not affect the data push. It's just that several lines of data are pushed at once. As long as the data sent at one time is complete, it's fine. Because the client parses one line at a time. As for the disorder issue of the term, the client can make its own judgment. Of course, the server can also make some appropriate optimizations.
> > 我认为并发往ChannelHandlerContext里面write数据是存在这种可能性，但是不影响数据推送，无非一下子推送了好几行数据，服务端只要确保一次性发出去的数据是完整的就行，因为客户端是一行行解析的。至于term的乱序问题 客户端可以自己判断，当然服务端也可以适当的做一点优化
>
> 这一块并发写入不会破坏数据完整性，不需要完全顺序处理，保持并发性能，只做必要的过滤优化 @funky-eyes wdyt?
>
> ```java
> // For HTTP/2, headers must be sent first on the initial response
> if (sendHeaders) {
> Http2Headers headers = new DefaultHttp2Headers().status(nettyStatus.codeAsText());
> headers.set(HttpHeaderNames.CONTENT_TYPE, "text/event-stream; charset=utf-8");
> headers.set(HttpHeaderNames.CACHE_CONTROL, "no-cache");
>
> ctx.write(new DefaultHttp2HeadersFrame(headers));
> }
>
> String group = watcher.getGroup();
> Long finalTerm = term != null ? term : GROUP_UPDATE_TERM.getOrDefault(group, 0L);
> String eventData = buildEventData(nettyStatus, closeStream, sendHeaders, group, finalTerm);
> ByteBuf content = Unpooled.copiedBuffer(eventData, StandardCharsets.UTF_8);
>
> // Send DATA frame (if closeStream is true, it will end the current stream)
> ctx.write(new DefaultHttp2DataFrame(content, closeStream));
> ctx.flush();
> ```
The out-of-order issue I mentioned isn't limited to HTTP/2 responses alone. The onChangeEvent method is annotated with @Async, and notifications to watchers are performed using parallelStream. As a result, when the cluster experiences a rapid series of consecutive changes, this could lead to out-of-order notifications (among other potential issues). Of course, addressing this can be treated as a separate optimization task.
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -57,24 +59,35 @@ public class ClusterWatcherManager implements ClusterChangeListener {
private static final Map<String, Long> GROUP_UPDATE_TERM = new ConcurrentHashMap<>();
+ private static final Map<Watcher<HttpContext>, Boolean> HTTP2_HEADERS_SENT = new ConcurrentHashMap<>();
+
private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor =
new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("long-polling", 1));
@PostConstruct
public void init() {
- // Responds to monitors that time out
+ // Periodically check and respond to watchers that have timed out
scheduledThreadPoolExecutor.scheduleAtFixedRate(
() -> {
for (String group : WATCHERS.keySet()) {
Optional.ofNullable(WATCHERS.remove(group))
.ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
- if (System.currentTimeMillis() >= watcher.getTimeout()) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.NOT_MODIFIED);
- }
- if (!watcher.isDone()) {
- // Re-register
- registryWatcher(watcher);
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
+ if (isHttp2) {
+ if (!context.getContext().channel().isActive()) {
+ watcher.setDone(true);
+ HTTP2_HEADERS_SENT.remove(watcher);
+ } else {
+ registryWatcher(watcher);
Review Comment:
> > I believe that writing data to the ChannelHandlerContext in a concurrent manner does have this possibility, but it does not affect the data push. It's just that several lines of data are pushed at once. As long as the data sent at one time is complete, it's fine. Because the client parses one line at a time. As for the disorder issue of the term, the client can make its own judgment. Of course, the server can also make some appropriate optimizations.
> > 我认为并发往ChannelHandlerContext里面write数据是存在这种可能性，但是不影响数据推送，无非一下子推送了好几行数据，服务端只要确保一次性发出去的数据是完整的就行，因为客户端是一行行解析的。至于term的乱序问题 客户端可以自己判断，当然服务端也可以适当的做一点优化
>
> 这一块并发写入不会破坏数据完整性，不需要完全顺序处理，保持并发性能，只做必要的过滤优化 @funky-eyes wdyt?
>
> ```java
> // For HTTP/2, headers must be sent first on the initial response
> if (sendHeaders) {
> Http2Headers headers = new DefaultHttp2Headers().status(nettyStatus.codeAsText());
> headers.set(HttpHeaderNames.CONTENT_TYPE, "text/event-stream; charset=utf-8");
> headers.set(HttpHeaderNames.CACHE_CONTROL, "no-cache");
>
> ctx.write(new DefaultHttp2HeadersFrame(headers));
> }
>
> String group = watcher.getGroup();
> Long finalTerm = term != null ? term : GROUP_UPDATE_TERM.getOrDefault(group, 0L);
> String eventData = buildEventData(nettyStatus, closeStream, sendHeaders, group, finalTerm);
> ByteBuf content = Unpooled.copiedBuffer(eventData, StandardCharsets.UTF_8);
>
> // Send DATA frame (if closeStream is true, it will end the current stream)
> ctx.write(new DefaultHttp2DataFrame(content, closeStream));
> ctx.flush();
> ```
The out-of-order issue I mentioned isn't limited to HTTP/2 responses alone. The onChangeEvent method is annotated with `@Async`, and notifications to watchers are performed using parallelStream. As a result, when the cluster experiences a rapid series of consecutive changes, this could lead to out-of-order notifications (among other potential issues). Of course, addressing this can be treated as a separate optimization task.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2777228789
##########
common/src/main/java/org/apache/seata/common/util/HttpClientUtil.java:
##########
@@ -265,25 +215,101 @@ private static String buildUrlWithParams(String url, Map<String, String> params)
return urlBuilder.toString();
}
- private static void executeAsync(OkHttpClient client, Request request, final HttpCallback<Response> callback) {
- client.newCall(request).enqueue(new Callback() {
- @Override
- public void onResponse(Call call, Response response) {
- try {
- callback.onSuccess(response);
- } finally {
- response.close();
- }
- }
+ /**
+ * Create an HTTP/2 client for watch connections.
+ * This client is configured for long-lived connections to receive Server-Sent Events (SSE).
+ * The client instances are cached and reused based on the connection timeout to improve performance.
+ *
+ * @param connectTimeoutSeconds connection timeout in seconds (fast failure if server is unreachable)
+ * @return configured OkHttpClient instance (cached and reused)
+ */
+ private static OkHttpClient createHttp2WatchClient(int connectTimeoutSeconds) {
+ return HTTP2_CLIENT_MAP.computeIfAbsent(connectTimeoutSeconds, k -> new OkHttpClient.Builder()
+ .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+ // Fast failure during connection phase
+ .connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ // Infinite read timeout to allow continuous listening for server push
+ .readTimeout(0, TimeUnit.SECONDS)
Review Comment:
done in [fafe65a](https://github.com/apache/incubator-seata/pull/7903/commits/fafe65ad35d1d19a8d1524cef8baa26ecf08e43c)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "YvCeung (via GitHub)" <gi...@apache.org>.
YvCeung commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2777211362
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -89,36 +141,104 @@ public void init() {
@Async
public void onChangeEvent(ClusterChangeEvent event) {
if (event.getTerm() > 0) {
- GROUP_UPDATE_TERM.put(event.getGroup(), event.getTerm());
- // Notifications are made of changes in cluster information
- Optional.ofNullable(WATCHERS.remove(event.getGroup()))
- .ifPresent(watchers -> watchers.parallelStream().forEach(this::notifyWatcher));
+ String group = event.getGroup();
+ Long eventTerm = event.getTerm();
+
+ Long currentTerm = GROUP_UPDATE_TERM.get(group);
+ if (currentTerm != null && eventTerm <= currentTerm) {
+ logger.info(
+ "Discarding outdated event with term {} for group {}, current term is {}",
+ eventTerm,
+ group,
+ currentTerm);
+ return;
+ }
+
+ GROUP_UPDATE_TERM.put(group, eventTerm);
+
+ // Handle HTTP/1.1 watchers: remove and notify (one-time request)
+ Optional.ofNullable(HTTP1_WATCHERS.remove(group))
+ .ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
+ notifyWatcher(watcher, eventTerm);
+ // HTTP/1.1 watcher is done after notification
+ watcher.setDone(true);
+ }));
+
+ // Handle HTTP/2 watchers: notify without removing (long-lived connection)
+ Queue<Watcher<HttpContext>> http2Watchers = HTTP2_WATCHERS.get(group);
+ if (http2Watchers != null && !http2Watchers.isEmpty()) {
+ List<Watcher<HttpContext>> watchersToNotify = new ArrayList<>(http2Watchers);
+ watchersToNotify.forEach(watcher -> {
+ if (watcher.getAsyncContext().getContext().channel().isActive() && !watcher.isDone()) {
+ if (eventTerm > watcher.getTerm()) {
+ notifyWatcher(watcher, eventTerm);
+ } else {
+ logger.info(
+ "Skipping notification for watcher with term {} >= event term {} for group {}",
+ watcher.getTerm(),
+ eventTerm,
+ group);
+ }
+ } else {
+ // Remove inactive watcher
+ http2Watchers.remove(watcher);
+ HTTP2_HEADERS_SENT.remove(watcher);
+ }
+ });
Review Comment:
我们是在“只读地遍历快照”，在需要时从原队列里按对象引用删除同一个 watcher。没有在遍历 http2Watchers 的同时修改它，所以不存在“边迭代原队列边删”的典型并发修改问题。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature:Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#issuecomment-3696953844
## [Codecov](https://app.codecov.io/gh/apache/incubator-seata/pull/7903?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `24.44444%` with `136 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 70.91%. Comparing base ([`b1db26b`](https://app.codecov.io/gh/apache/incubator-seata/commit/b1db26b0221d909a724d6b734cfcc86c07838181?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`467fc38`](https://app.codecov.io/gh/apache/incubator-seata/commit/467fc38095c81c5da9bc5152c0ddfa5f1e2bcacd?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/incubator-seata/pull/7903?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...a/org/apache/seata/common/util/SeataHttpWatch.java](https://app.codecov.io/gh/apache/incubator-seata/pull/7903?src=pr&el=tree&filepath=common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fseata%2Fcommon%2Futil%2FSeataHttpWatch.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9zZWF0YS9jb21tb24vdXRpbC9TZWF0YUh0dHBXYXRjaC5qYXZh) | 0.00% | [80 Missing :warning: ](https://app.codecov.io/gh/apache/incubator-seata/pull/7903?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...a/org/apache/seata/common/util/HttpClientUtil.java](https://app.codecov.io/gh/apache/incubator-seata/pull/7903?src=pr&el=tree&filepath=common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fseata%2Fcommon%2Futil%2FHttpClientUtil.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9zZWF0YS9jb21tb24vdXRpbC9IdHRwQ2xpZW50VXRpbC5qYXZh) | 0.00% | [31 Missing :warning: ](https://app.codecov.io/gh/apache/incubator-seata/pull/7903?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...pache/seata/common/metadata/ClusterWatchEvent.java](https://app.codecov.io/gh/apache/incubator-seata/pull/7903?src=pr&el=tree&filepath=common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fseata%2Fcommon%2Fmetadata%2FClusterWatchEvent.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9zZWF0YS9jb21tb24vbWV0YWRhdGEvQ2x1c3RlcldhdGNoRXZlbnQuamF2YQ==) | 0.00% | [14 Missing :warning: ](https://app.codecov.io/gh/apache/incubator-seata/pull/7903?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../server/cluster/manager/ClusterWatcherManager.java](https://app.codecov.io/gh/apache/incubator-seata/pull/7903?src=pr&el=tree&filepath=server%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fseata%2Fserver%2Fcluster%2Fmanager%2FClusterWatcherManager.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9zZWF0YS9zZXJ2ZXIvY2x1c3Rlci9tYW5hZ2VyL0NsdXN0ZXJXYXRjaGVyTWFuYWdlci5qYXZh) | 80.00% | [3 Missing and 8 partials :warning: ](https://app.codecov.io/gh/apache/incubator-seata/pull/7903?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## 2.x #7903 +/- ##
============================================
- Coverage 71.10% 70.91% -0.20%
Complexity 797 797
============================================
Files 1294 1296 +2
Lines 49538 49701 +163
Branches 5879 5908 +29
============================================
+ Hits 35222 35243 +21
- Misses 11406 11544 +138
- Partials 2910 2914 +4
```
| [Files with missing lines](https://app.codecov.io/gh/apache/incubator-seata/pull/7903?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [.../server/cluster/manager/ClusterWatcherManager.java](https://app.codecov.io/gh/apache/incubator-seata/pull/7903?src=pr&el=tree&filepath=server%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fseata%2Fserver%2Fcluster%2Fmanager%2FClusterWatcherManager.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9zZWF0YS9zZXJ2ZXIvY2x1c3Rlci9tYW5hZ2VyL0NsdXN0ZXJXYXRjaGVyTWFuYWdlci5qYXZh) | `81.44% <80.00%> (-1.61%)` | :arrow_down: |
| [...pache/seata/common/metadata/ClusterWatchEvent.java](https://app.codecov.io/gh/apache/incubator-seata/pull/7903?src=pr&el=tree&filepath=common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fseata%2Fcommon%2Fmetadata%2FClusterWatchEvent.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9zZWF0YS9jb21tb24vbWV0YWRhdGEvQ2x1c3RlcldhdGNoRXZlbnQuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...a/org/apache/seata/common/util/HttpClientUtil.java](https://app.codecov.io/gh/apache/incubator-seata/pull/7903?src=pr&el=tree&filepath=common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fseata%2Fcommon%2Futil%2FHttpClientUtil.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9zZWF0YS9jb21tb24vdXRpbC9IdHRwQ2xpZW50VXRpbC5qYXZh) | `69.71% <0.00%> (-15.01%)` | :arrow_down: |
| [...a/org/apache/seata/common/util/SeataHttpWatch.java](https://app.codecov.io/gh/apache/incubator-seata/pull/7903?src=pr&el=tree&filepath=common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fseata%2Fcommon%2Futil%2FSeataHttpWatch.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9zZWF0YS9jb21tb24vdXRpbC9TZWF0YUh0dHBXYXRjaC5qYXZh) | `0.00% <0.00%> (ø)` | |
... and [9 files with indirect coverage changes](https://app.codecov.io/gh/apache/incubator-seata/pull/7903/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
[](https://app.codecov.io/gh/apache/incubator-seata/pull/7903?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org
Re: [PR] feature: Support HTTP/2 stream push for the Watch API in Server Raft mode [incubator-seata]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #7903:
URL: https://github.com/apache/incubator-seata/pull/7903#discussion_r2659250081
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -57,24 +59,35 @@ public class ClusterWatcherManager implements ClusterChangeListener {
private static final Map<String, Long> GROUP_UPDATE_TERM = new ConcurrentHashMap<>();
+ private static final Map<Watcher<HttpContext>, Boolean> HTTP2_HEADERS_SENT = new ConcurrentHashMap<>();
+
private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor =
new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("long-polling", 1));
@PostConstruct
public void init() {
- // Responds to monitors that time out
+ // Periodically check and respond to watchers that have timed out
scheduledThreadPoolExecutor.scheduleAtFixedRate(
() -> {
for (String group : WATCHERS.keySet()) {
Optional.ofNullable(WATCHERS.remove(group))
.ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
- if (System.currentTimeMillis() >= watcher.getTimeout()) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.NOT_MODIFIED);
- }
- if (!watcher.isDone()) {
- // Re-register
- registryWatcher(watcher);
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
+ if (isHttp2) {
+ if (!context.getContext().channel().isActive()) {
+ watcher.setDone(true);
+ HTTP2_HEADERS_SENT.remove(watcher);
+ } else {
+ registryWatcher(watcher);
+ }
+ } else {
+ if (System.currentTimeMillis() >= watcher.getTimeout()) {
+ watcher.setDone(true);
+ sendWatcherResponse(watcher, HttpResponseStatus.NOT_MODIFIED, true, false);
+ } else if (!watcher.isDone()) {
+ registryWatcher(watcher);
+ }
}
}));
}
Review Comment:
The scheduled task removes all watchers from a group at once using `WATCHERS.remove(group)`, then processes them. If new watchers are registered for the same group during processing, they could be lost. Consider using a more concurrent-safe approach, such as iterating over watchers without removing the entire queue first, or using a copy-on-write approach.
##########
server/src/test/java/org/apache/seata/server/controller/ClusterControllerTest.java:
##########
@@ -81,43 +76,96 @@ void watchTimeoutTest() throws Exception {
@Test
@Order(2)
- void watchTimeoutTest_withHttp2() throws Exception {
- CountDownLatch latch = new CountDownLatch(1);
-
+ void watchTimeoutTest_http2() throws Exception {
Map<String, String> headers = new HashMap<>();
- headers.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
+ headers.put("Content-Type", ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
Map<String, String> params = new HashMap<>();
params.put("default-test", "1");
- HttpCallback<Response> callback = new HttpCallback<Response>() {
- @Override
- public void onSuccess(Response response) {
- Assertions.assertNotNull(response);
- Assertions.assertEquals(Protocol.H2_PRIOR_KNOWLEDGE, response.protocol());
- Assertions.assertEquals(HttpStatus.SC_NOT_MODIFIED, response.code());
- latch.countDown();
+ // For HTTP2, the connection should remain open and not timeout
+ // The test verifies that the connection stays alive beyond the timeout period
+ try (SeataHttpWatch<ClusterWatchEvent> watch = HttpClientUtil.watchPost(
+ "http://127.0.0.1:" + port + "/metadata/v1/watch?timeout=3000",
+ params,
+ headers,
+ ClusterWatchEvent.class)) {
+
+ boolean keepaliveReceived = false;
+ long startTime = System.currentTimeMillis();
+ // Verify KEEPALIVE is received immediately
+ boolean hasNext = watch.hasNext();
+ if (!hasNext) {
+ Assertions.fail("Expect KEEPALIVE Event Received...");
}
-
- @Override
- public void onFailure(Throwable t) {
- Assertions.fail("Should not fail");
+ SeataHttpWatch.Response<ClusterWatchEvent> firstResponse = watch.next();
+ Assertions.assertEquals(
+ SeataHttpWatch.Response.Type.KEEPALIVE, firstResponse.type, "First event should be KEEPALIVE");
+ Assertions.assertNotNull(firstResponse.object, "Event data should not be null");
+ ClusterWatchEvent keepaliveEvent = firstResponse.object;
+ Assertions.assertEquals("keepalive", keepaliveEvent.getType(), "Event type should be 'keepalive'");
+ Assertions.assertEquals("default-test", keepaliveEvent.getGroup(), "Group should match");
+ Assertions.assertNotNull(keepaliveEvent.getTimestamp(), "Timestamp should not be null");
+ keepaliveReceived = true;
+
+ long elapsed = System.currentTimeMillis() - startTime;
+ Assertions.assertTrue(
+ elapsed < 1000,
+ "KEEPALIVE should be received immediately after connection, elapsed: " + elapsed + "ms");
+ Assertions.assertTrue(keepaliveReceived, "Keepalive should be received after connection");
+
+ // Verify connection remains alive beyond timeout period (3000ms)
+ // Note: hasNext() blocks when no data is available, so we cannot use it to check connection status
+ // Instead, we wait beyond the timeout period and then trigger a cluster change event
+ // If we can receive the event, it proves the connection is still alive
+ long timeoutPeriod = 3000; // Timeout period from query parameter
+ long waitTime = timeoutPeriod + 1000; // Wait timeout period + 1 second
+ try {
+ Thread.sleep(waitTime);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ Assertions.fail("Test interrupted while waiting for connection verification");
}
- @Override
- public void onCancelled() {
- Assertions.fail("Should not be cancelled");
+ // Trigger a cluster change event to verify connection is still active
+ Thread triggerThread = new Thread(() -> {
+ try {
+ Thread.sleep(2000); // Small delay to ensure we're past timeout
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ ((ApplicationEventPublisher) ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT))
+ .publishEvent(new ClusterChangeEvent(this, "default-test", 2, true));
+ });
+ triggerThread.start();
+ boolean clusterUpdateReceived = false;
+ if (!watch.hasNext()) {
+ Assertions.fail("Expect cluster-update Event Received...");
+ }
Review Comment:
The test starts a trigger thread that sleeps for 2000ms, but immediately after starting the thread it calls `watch.hasNext()` which will block. This creates a race condition where `hasNext()` might block indefinitely if the trigger thread hasn't published the event yet. The thread should be started before the sleep on line 124, or the wait logic should be restructured to ensure the trigger thread completes before checking for the next event.
##########
common/src/main/java/org/apache/seata/common/util/SeataHttpWatch.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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.seata.common.util;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import okhttp3.Call;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.ResponseBody;
+import okio.BufferedSource;
+import org.apache.seata.common.exception.FrameworkException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+/**
+ * Seata HTTP/2 Watch implementation.
+ * Consumes server-pushed SSE data frames via an iterator-style API.
+ *
+ * @param <T> event data type
+ */
+public class SeataHttpWatch<T>
+ implements Iterator<SeataHttpWatch.Response<T>>, Iterable<SeataHttpWatch.Response<T>>, AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SeataHttpWatch.class);
+
+ private final ResponseBody responseBody;
+ private final BufferedSource source;
+ private final Call call;
+ private final ObjectMapper objectMapper;
+ private final Class<T> eventType;
+
+ /**
+ * Response wrapper containing event type and data
+ */
+ public static class Response<T> {
+ /**
+ * Event type enum - matches SSE event field values
+ */
+ public enum Type {
+ /**
+ * Cluster update event
+ */
+ CLUSTER_UPDATE,
+ /**
+ * Keep-alive event (connection established)
+ */
+ KEEPALIVE,
+ /**
+ * Timeout event (stream closed due to timeout)
+ */
+ TIMEOUT,
+ /**
+ * Error event
+ */
+ ERROR
+ }
+
+ public final Type type;
+
+ public final T object;
+
+ public Response(Type type, T object) {
+ this.type = type;
+ this.object = object;
+ }
+ }
+
+ /**
+ * Create a Watch instance from an OkHttp call
+ *
+ * @param call the prepared HTTP call
+ * @param eventType the class type for deserializing event data
+ * @param <T> the event data type
+ * @return a Watch instance
+ * @throws IOException if the request fails
+ */
+ public static <T> SeataHttpWatch<T> createWatch(Call call, Class<T> eventType) throws IOException {
+
+ okhttp3.Response response = call.execute();
+
+ if (!response.isSuccessful()) {
+ String respBody = null;
+ try (ResponseBody body = response.body()) {
+ if (body != null) {
+ respBody = body.string();
+ }
+ } catch (IOException e) {
+ throw new FrameworkException(e, "Watch request failed: " + response.message());
+ }
+ throw new FrameworkException(
+ String.format("Watch request failed with code %d: %s", response.code(), respBody));
+ }
+
+ // Verify Content-Type is SSE
+ String contentType = response.header("Content-Type");
+ if (contentType == null || !contentType.contains("text/event-stream")) {
+ LOGGER.warn("Expected Content-Type: text/event-stream, got: {}", contentType);
+ }
+
+ return new SeataHttpWatch<>(response.body(), call, eventType);
+ }
+
+ /**
+ * Create a Watch instance with a prepared request
+ *
+ * @param client the OkHttpClient instance
+ * @param request the HTTP request
+ * @param eventType the class type for deserializing event data
+ * @param <T> the event data type
+ * @return a Watch instance
+ * @throws IOException if the request fails
+ */
+ public static <T> SeataHttpWatch<T> createWatch(OkHttpClient client, Request request, Class<T> eventType)
+ throws IOException {
+
+ Call call = client.newCall(request);
+ return createWatch(call, eventType);
+ }
+
+ private SeataHttpWatch(ResponseBody responseBody, Call call, Class<T> eventType) {
+ this.responseBody = responseBody;
+ this.source = responseBody.source();
+ this.call = call;
+ this.objectMapper = new ObjectMapper();
+ this.eventType = eventType;
+ }
+
+ @Override
+ public boolean hasNext() {
+ try {
+ // Check if source is exhausted (stream closed)
+ return !source.exhausted();
+ } catch (IOException e) {
+ LOGGER.error("Error checking if stream has more data", e);
+ return false;
+ }
+ }
+
+ @Override
+ public Response<T> next() {
+ try {
+ // Read complete SSE event in a loop (not recursive to avoid stack overflow)
+ // SSE format: "data: {json}\n\n"
+ // Each event is separated by double newline
+ // Event type is included in the JSON data, not in a separate "event:" field
+
+ StringBuilder dataBuffer = new StringBuilder();
+
+ // Read lines until we get a complete event (ending with empty line)
+ // This loop reads all lines of a single event atomically
+ while (true) {
+ String line = source.readUtf8Line();
+
+ if (line == null) {
+ // Stream closed
+ if (dataBuffer.length() > 0) {
+ // We have partial data, try to parse it
+ String eventJson = dataBuffer.toString();
+ return parseEvent(eventJson);
+ }
+ throw new RuntimeException("Stream closed unexpectedly");
+ }
+
+ if (line.startsWith("data: ")) {
+ String jsonData = line.substring(6);
+ if (dataBuffer.length() > 0) {
+ dataBuffer.append('\n');
+ }
+ dataBuffer.append(jsonData);
+ } else if (line.isEmpty()) {
+ // Empty line indicates end of event
+ if (dataBuffer.length() > 0) {
+ String eventJson = dataBuffer.toString();
+ return parseEvent(eventJson);
+ }
+ // Empty line but no data, continue reading (skip blank lines between events)
+ } else {
+ LOGGER.debug("Unknown SSE line format, ignoring: {}", line);
+ }
+ }
+
+ } catch (IOException e) {
+ throw new RuntimeException("IO Exception during next()", e);
+ }
+ }
+
+ /**
+ * Parse SSE event JSON into Response object
+ *
+ * @param json the JSON string to parse (contains type field)
+ * @return the parsed Response object
+ * @throws IOException if parsing fails
+ */
+ private Response<T> parseEvent(String json) throws IOException {
+ try {
+ // Parse JSON once to get both event type and full data
+ // First, parse to extract event type
+ EventMetadata metadata = objectMapper.readValue(json, EventMetadata.class);
+
+ // Map event type from JSON to Response.Type
+ Response.Type responseType = mapEventTypeToResponseType(metadata.type);
+
+ // Deserialize the full event data (parse once, reuse if possible)
+ T eventData = objectMapper.readValue(json, eventType);
Review Comment:
The JSON is being parsed twice - once to extract the event type (line 216) and again to deserialize the full event data (line 222). This is inefficient and could impact performance, especially with high-frequency events. Consider parsing once and extracting the type from the deserialized object, or using a single pass with a custom deserializer.
##########
server/src/test/java/org/apache/seata/server/controller/ClusterControllerTest.java:
##########
@@ -138,63 +186,176 @@ public void run() {
try (Response response =
HttpClientUtil.doPost("http://127.0.0.1:" + port + "/metadata/v1/watch", param, header, 30000)) {
if (response != null) {
- Assertions.assertEquals(HttpStatus.SC_OK, response.code());
+ Assertions.assertEquals(200, response.code());
return;
}
}
Assertions.fail();
}
@Test
- @Order(5)
- void watch_withHttp2() throws Exception {
- CountDownLatch latch = new CountDownLatch(1);
-
- Map<String, String> headers = new HashMap<>();
- headers.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
-
- Map<String, String> params = new HashMap<>();
- params.put("default-test", "1");
-
- Thread thread = new Thread(() -> {
- try {
- Thread.sleep(2000);
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
+ @Order(4)
+ void watchStream_http2() throws Exception {
+ Map<String, String> header = new HashMap<>();
+ header.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
+ Map<String, String> param = new HashMap<>();
+ param.put("default-test", "1");
+ Thread thread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ ((ApplicationEventPublisher) ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT))
+ .publishEvent(new ClusterChangeEvent(this, "default-test", 2, true));
}
- ((ApplicationEventPublisher) ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT))
- .publishEvent(new ClusterChangeEvent(this, "default-test", 2, true));
});
thread.start();
- HttpCallback<Response> callback = new HttpCallback<Response>() {
- @Override
- public void onSuccess(Response response) {
- Assertions.assertNotNull(response);
- Assertions.assertEquals(Protocol.H2_PRIOR_KNOWLEDGE, response.protocol());
- Assertions.assertEquals(HttpStatus.SC_OK, response.code());
- latch.countDown();
- }
+ boolean keepaliveReceived = false;
+ boolean clusterUpdateReceived = false;
+ long startTime = System.currentTimeMillis();
+ long maxWaitTime = 10000; // Maximum wait time: 10 seconds
- @Override
- public void onFailure(Throwable t) {
- Assertions.fail("Should not fail: " + t.getMessage());
+ try (SeataHttpWatch<ClusterWatchEvent> watch = HttpClientUtil.watchPost(
+ "http://127.0.0.1:" + port + "/metadata/v1/watch", param, header, ClusterWatchEvent.class)) {
+ // For HTTP2, connection will remain open, so we need to break after receiving expected events
+ // Note: hasNext() blocks when no data is available, so we use next() directly with timeout protection
+ while (System.currentTimeMillis() - startTime < maxWaitTime) {
+ try {
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
Review Comment:
The `watch.next()` call blocks indefinitely if no data is available. The timeout protection using `System.currentTimeMillis() - startTime < maxWaitTime` only checks the elapsed time but doesn't interrupt a blocking `next()` call. If the server never sends an event, this test could hang for much longer than 10 seconds. Consider using a separate thread with timeout or a non-blocking approach to read events.
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -57,24 +59,35 @@ public class ClusterWatcherManager implements ClusterChangeListener {
private static final Map<String, Long> GROUP_UPDATE_TERM = new ConcurrentHashMap<>();
+ private static final Map<Watcher<HttpContext>, Boolean> HTTP2_HEADERS_SENT = new ConcurrentHashMap<>();
+
private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor =
new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("long-polling", 1));
@PostConstruct
public void init() {
- // Responds to monitors that time out
+ // Periodically check and respond to watchers that have timed out
scheduledThreadPoolExecutor.scheduleAtFixedRate(
() -> {
for (String group : WATCHERS.keySet()) {
Optional.ofNullable(WATCHERS.remove(group))
.ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
- if (System.currentTimeMillis() >= watcher.getTimeout()) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.NOT_MODIFIED);
- }
- if (!watcher.isDone()) {
- // Re-register
- registryWatcher(watcher);
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
Review Comment:
The check `context instanceof HttpContext` is redundant since the parameter type already declares `context` as `HttpContext`. This check will always return true and should be removed.
```suggestion
boolean isHttp2 = context.isHttp2();
```
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -128,27 +174,63 @@ private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatu
} else {
ctx.writeAndFlush(response);
}
- } else {
- // HTTP/2 response (h2c support)
- // Send headers frame
+ return;
+ }
+
+ // For HTTP/2, headers must be sent first on the initial response
+ if (sendHeaders) {
Http2Headers headers = new DefaultHttp2Headers().status(nettyStatus.codeAsText());
- headers.set(HttpHeaderNames.CONTENT_LENGTH, "0");
+ headers.set(HttpHeaderNames.CONTENT_TYPE, "text/event-stream; charset=utf-8");
+ headers.set(HttpHeaderNames.CACHE_CONTROL, "no-cache");
+
ctx.write(new DefaultHttp2HeadersFrame(headers));
+ }
+
+ String group = watcher.getGroup();
+ String sse = buildSSEFormat(nettyStatus, closeStream, sendHeaders, group);
+
+ ByteBuf content = Unpooled.copiedBuffer(sse, StandardCharsets.UTF_8);
- // Send empty data frame with endStream=true to close the stream
- ctx.writeAndFlush(new DefaultHttp2DataFrame(Unpooled.EMPTY_BUFFER, true))
- .addListener(f -> {
- if (!f.isSuccess()) {
- logger.warn("HTTP2 response send failed, group={}", group, f.cause());
- }
- });
+ // Send DATA frame (if closeStream is true, it will end the current stream)
+ ctx.write(new DefaultHttp2DataFrame(content, closeStream));
+ ctx.flush();
+ }
+
+ private String buildSSEFormat(
+ HttpResponseStatus nettyStatus, boolean closeStream, boolean sendHeaders, String group) {
+ // Determine event type (embedded in JSON, not in SSE event field)
+ String eventType;
+ if (sendHeaders) {
+ // Send keepalive event when stream is first established to confirm connection
+ eventType = "keepalive";
+ } else if (closeStream && nettyStatus == HttpResponseStatus.NOT_MODIFIED) {
+ // Timeout event, stream needs to be closed
+ eventType = "timeout";
+ } else {
+ // Normal cluster update event
+ eventType = "cluster-update";
}
+
+ String json = String.format(
+ "{\"type\":\"%s\",\"group\":\"%s\",\"term\":%d,\"timestamp\":%d}",
+ eventType, group, GROUP_UPDATE_TERM.getOrDefault(group, 0L), System.currentTimeMillis());
+ logger.debug("Sending watch event: {}", json);
+
+ // SSE format: only send data: field, event type is embedded in JSON
+ return "data: " + json + "\n\n";
}
public void registryWatcher(Watcher<HttpContext> watcher) {
String group = watcher.getGroup();
Long term = GROUP_UPDATE_TERM.get(group);
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
Review Comment:
The check `context instanceof HttpContext` is redundant since the parameter type already declares `context` as `HttpContext`. This check will always return true and should be removed.
```suggestion
boolean isHttp2 = context.isHttp2();
```
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -90,35 +103,68 @@ public void init() {
public void onChangeEvent(ClusterChangeEvent event) {
if (event.getTerm() > 0) {
GROUP_UPDATE_TERM.put(event.getGroup(), event.getTerm());
- // Notifications are made of changes in cluster information
+ // Notify all watchers of cluster information changes
Optional.ofNullable(WATCHERS.remove(event.getGroup()))
.ifPresent(watchers -> watchers.parallelStream().forEach(this::notifyWatcher));
}
}
private void notifyWatcher(Watcher<HttpContext> watcher) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.OK);
- }
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
Review Comment:
The check `context instanceof HttpContext` is redundant since the parameter type already declares `context` as `HttpContext`. This check will always return true and should be removed.
##########
server/src/test/java/org/apache/seata/server/controller/ClusterControllerTest.java:
##########
@@ -138,63 +186,176 @@ public void run() {
try (Response response =
HttpClientUtil.doPost("http://127.0.0.1:" + port + "/metadata/v1/watch", param, header, 30000)) {
if (response != null) {
- Assertions.assertEquals(HttpStatus.SC_OK, response.code());
+ Assertions.assertEquals(200, response.code());
return;
}
}
Assertions.fail();
}
@Test
- @Order(5)
- void watch_withHttp2() throws Exception {
- CountDownLatch latch = new CountDownLatch(1);
-
- Map<String, String> headers = new HashMap<>();
- headers.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
-
- Map<String, String> params = new HashMap<>();
- params.put("default-test", "1");
-
- Thread thread = new Thread(() -> {
- try {
- Thread.sleep(2000);
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
+ @Order(4)
+ void watchStream_http2() throws Exception {
+ Map<String, String> header = new HashMap<>();
+ header.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
+ Map<String, String> param = new HashMap<>();
+ param.put("default-test", "1");
+ Thread thread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ ((ApplicationEventPublisher) ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT))
+ .publishEvent(new ClusterChangeEvent(this, "default-test", 2, true));
}
- ((ApplicationEventPublisher) ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT))
- .publishEvent(new ClusterChangeEvent(this, "default-test", 2, true));
});
thread.start();
- HttpCallback<Response> callback = new HttpCallback<Response>() {
- @Override
- public void onSuccess(Response response) {
- Assertions.assertNotNull(response);
- Assertions.assertEquals(Protocol.H2_PRIOR_KNOWLEDGE, response.protocol());
- Assertions.assertEquals(HttpStatus.SC_OK, response.code());
- latch.countDown();
- }
+ boolean keepaliveReceived = false;
+ boolean clusterUpdateReceived = false;
+ long startTime = System.currentTimeMillis();
+ long maxWaitTime = 10000; // Maximum wait time: 10 seconds
- @Override
- public void onFailure(Throwable t) {
- Assertions.fail("Should not fail: " + t.getMessage());
+ try (SeataHttpWatch<ClusterWatchEvent> watch = HttpClientUtil.watchPost(
+ "http://127.0.0.1:" + port + "/metadata/v1/watch", param, header, ClusterWatchEvent.class)) {
+ // For HTTP2, connection will remain open, so we need to break after receiving expected events
+ // Note: hasNext() blocks when no data is available, so we use next() directly with timeout protection
+ while (System.currentTimeMillis() - startTime < maxWaitTime) {
+ try {
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
+ SeataHttpWatch.Response.Type type = response.type;
+
+ if (type == SeataHttpWatch.Response.Type.KEEPALIVE) {
+ keepaliveReceived = true;
+ Assertions.assertNotNull(response.object, "KEEPALIVE event data should not be null");
+ Assertions.assertEquals(
+ "keepalive", response.object.getType(), "Event type should be 'keepalive'");
+ } else if (type == SeataHttpWatch.Response.Type.CLUSTER_UPDATE) {
+ clusterUpdateReceived = true;
+ Assertions.assertNotNull(response.object, "CLUSTER_UPDATE event data should not be null");
+ Assertions.assertEquals(
+ "cluster-update", response.object.getType(), "Event type should be 'cluster-update'");
+ Assertions.assertEquals("default-test", response.object.getGroup(), "Group should match");
+ Assertions.assertNotNull(response.object.getTerm(), "Term should not be null");
+ Assertions.assertEquals(2L, response.object.getTerm().longValue(), "Term should be 2");
+ // After receiving cluster update, exit the loop
+ break;
+ }
+ } catch (Exception e) {
+ // If connection was closed or timeout, break the loop
+ if (keepaliveReceived && clusterUpdateReceived) {
+ // We already received expected events, break normally
+ break;
+ }
+ throw new RuntimeException("Unexpected exception while waiting for events", e);
+ }
}
- @Override
- public void onCancelled() {
- Assertions.fail("Should not be cancelled");
+ // Verify both events were received
+ Assertions.assertTrue(keepaliveReceived, "KEEPALIVE event should be received");
+ Assertions.assertTrue(
+ clusterUpdateReceived, "CLUSTER_UPDATE event should be received after cluster change");
+ } catch (IOException e) {
+ throw new RuntimeException("Watch failed", e);
+ }
+ }
+
+ @Test
+ @Order(5)
+ void watchMultipleClusterUpdates_http2() throws Exception {
+ Map<String, String> header = new HashMap<>();
+ header.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
+ Map<String, String> param = new HashMap<>();
+ param.put("default-test", "1");
+
+ // Trigger multiple cluster change events with different terms
+ Thread triggerThread = new Thread(() -> {
+ try {
+ Thread.sleep(3000); // Wait for connection to be established
+ ApplicationEventPublisher publisher = (ApplicationEventPublisher)
+ ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT);
+
+ // Trigger first cluster change event (term = 2)
+ publisher.publishEvent(new ClusterChangeEvent(this, "default-test", 2, true));
+ Thread.sleep(500);
+
+ // Trigger second cluster change event (term = 3)
+ publisher.publishEvent(new ClusterChangeEvent(this, "default-test", 3, true));
+ Thread.sleep(500);
+
+ // Trigger third cluster change event (term = 4)
+ publisher.publishEvent(new ClusterChangeEvent(this, "default-test", 4, true));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ triggerThread.start();
+
+ boolean keepaliveReceived = false;
+ int clusterUpdateCount = 0;
+ long startTime = System.currentTimeMillis();
+ long maxWaitTime = 10000; // Maximum wait time: 10 seconds
+ long[] expectedTerms = {2L, 3L, 4L};
+
+ try (SeataHttpWatch<ClusterWatchEvent> watch = HttpClientUtil.watchPost(
+ "http://127.0.0.1:" + port + "/metadata/v1/watch", param, header, ClusterWatchEvent.class)) {
+ // Receive events until we get all expected cluster updates
+ while (System.currentTimeMillis() - startTime < maxWaitTime && clusterUpdateCount < expectedTerms.length) {
+ try {
+ if (watch.hasNext()) {
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
+ SeataHttpWatch.Response.Type type = response.type;
+ if (type == SeataHttpWatch.Response.Type.KEEPALIVE) {
+ keepaliveReceived = true;
+ Assertions.assertNotNull(response.object, "KEEPALIVE event data should not be null");
+ Assertions.assertEquals(
+ "keepalive", response.object.getType(), "Event type should be 'keepalive'");
+ } else if (type == SeataHttpWatch.Response.Type.CLUSTER_UPDATE) {
+ clusterUpdateCount++;
+ Assertions.assertNotNull(response.object, "CLUSTER_UPDATE event data should not be null");
+ Assertions.assertEquals(
+ "cluster-update",
+ response.object.getType(),
+ "Event type should be 'cluster-update'");
+ Assertions.assertEquals("default-test", response.object.getGroup(), "Group should match");
+ Assertions.assertNotNull(response.object.getTerm(), "Term should not be null");
+
+ // Verify term matches expected value
+ long expectedTerm = expectedTerms[clusterUpdateCount - 1];
+ long actualTerm = response.object.getTerm().longValue();
+ Assertions.assertEquals(
+ expectedTerm,
+ actualTerm,
+ "Term should be " + expectedTerm + "But actualTerm is " + actualTerm
Review Comment:
The error message has a typo: "But actualTerm is" should be "but actualTerm is" (lowercase 'b'). Additionally, a space is missing before "But" in the concatenated string.
```suggestion
"Term should be " + expectedTerm + " but actualTerm is " + actualTerm
```
##########
common/src/main/java/org/apache/seata/common/util/HttpClientUtil.java:
##########
@@ -265,25 +215,101 @@ private static String buildUrlWithParams(String url, Map<String, String> params)
return urlBuilder.toString();
}
- private static void executeAsync(OkHttpClient client, Request request, final HttpCallback<Response> callback) {
- client.newCall(request).enqueue(new Callback() {
- @Override
- public void onResponse(Call call, Response response) {
- try {
- callback.onSuccess(response);
- } finally {
- response.close();
- }
- }
+ /**
+ * Create an HTTP/2 client for watch connections.
+ * This client is configured for long-lived connections to receive Server-Sent Events (SSE).
+ * The client instances are cached and reused based on the connection timeout to improve performance.
+ *
+ * @param connectTimeoutSeconds connection timeout in seconds (fast failure if server is unreachable)
+ * @return configured OkHttpClient instance (cached and reused)
+ */
+ private static OkHttpClient createHttp2WatchClient(int connectTimeoutSeconds) {
+ return HTTP2_CLIENT_MAP.computeIfAbsent(connectTimeoutSeconds, k -> new OkHttpClient.Builder()
+ .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+ // Fast failure during connection phase
+ .connectTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ // Infinite read timeout to allow continuous listening for server push
+ .readTimeout(0, TimeUnit.SECONDS)
+ .writeTimeout(connectTimeoutSeconds, TimeUnit.SECONDS)
+ .build());
+ }
- @Override
- public void onFailure(Call call, IOException e) {
- if (call.isCanceled()) {
- callback.onCancelled();
- } else {
- callback.onFailure(e);
- }
- }
- });
+ public static <T> SeataHttpWatch<T> watch(String url, Map<String, String> headers, Class<T> eventType)
+ throws IOException {
+ return watch(url, headers, null, "GET", eventType);
+ }
+
+ public static <T> SeataHttpWatch<T> watch(String url, Class<T> eventType) throws IOException {
+ return watch(url, null, null, "GET", eventType);
+ }
+
+ /**
+ * Execute a watch request with specified HTTP method and return a Watch iterator.
+ * This method creates a long-lived HTTP/2 connection to receive Server-Sent Events (SSE).
+ *
+ * @param url the URL to watch (must not be null or blank)
+ * @param headers HTTP headers
+ * @param requestBody request body (optional)
+ * @param method HTTP method (GET, POST, PUT)
+ * @param eventType the class type for deserializing event data
+ * @param <T> the event data type
+ * @return a Watch instance for receiving SSE events
+ * @throws IOException if the request fails
+ * @throws IllegalArgumentException if the URL is null or blank
+ */
+ private static <T> SeataHttpWatch<T> watch(
+ String url, Map<String, String> headers, RequestBody requestBody, String method, Class<T> eventType)
+ throws IOException {
+
+ if (StringUtils.isBlank(url)) {
+ throw new IllegalArgumentException("URL must not be null or blank");
+ }
+
+ OkHttpClient client = createHttp2WatchClient(HTTP2_WATCH_CONNECT_TIMEOUT_SECONDS);
+ Request request = buildHttp2WatchRequest(url, headers, requestBody, method);
+ return SeataHttpWatch.createWatch(client, request, eventType);
+ }
+
+ public static <T> SeataHttpWatch<T> watchPost(
+ String url, Map<String, String> params, Map<String, String> headers, Class<T> eventType)
+ throws IOException {
+ try {
+ String contentType = headers != null ? headers.get("Content-Type") : "";
+ RequestBody requestBody = createRequestBody(params, contentType);
+ return watch(url, headers, requestBody, "POST", eventType);
+ } catch (JsonProcessingException e) {
+ LOGGER.error("Failed to create request body", e);
+ throw new IOException("Failed to create request body", e);
+ }
+ }
+
+ public static <T> SeataHttpWatch<T> watchPost(String url, Map<String, String> params, Class<T> eventType)
+ throws IOException {
+ return watchPost(url, params, null, eventType);
+ }
+
+ private static Request buildHttp2WatchRequest(
+ String url, Map<String, String> headers, RequestBody requestBody, String method) {
+ Headers.Builder headerBuilder = new Headers.Builder();
+ if (headers != null) {
+ headers.forEach(headerBuilder::add);
+ }
+ // Always add Accept header for SSE
+ headerBuilder.add("Accept", "text/event-stream");
+
+ Request.Builder requestBuilder = new Request.Builder().url(url).headers(headerBuilder.build());
+
+ if ("POST".equals(method) && requestBody != null) {
+ requestBuilder.post(requestBody);
+ } else if ("PUT".equals(method) && requestBody != null) {
+ requestBuilder.put(requestBody);
+ } else if ("GET".equals(method)) {
+ requestBuilder.get();
+ } else {
+ // Default to GET if method is not specified or not supported
+ requestBuilder.get();
Review Comment:
The method allows POST and PUT with requestBody, but silently defaults to GET for any unsupported method (including DELETE, PATCH, etc.). This could mask errors when an incorrect method is specified. Consider throwing an IllegalArgumentException for unsupported methods instead of silently defaulting to GET.
```suggestion
throw new IllegalArgumentException("Unsupported HTTP method for watch request: " + method);
```
##########
server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java:
##########
@@ -90,35 +103,68 @@ public void init() {
public void onChangeEvent(ClusterChangeEvent event) {
if (event.getTerm() > 0) {
GROUP_UPDATE_TERM.put(event.getGroup(), event.getTerm());
- // Notifications are made of changes in cluster information
+ // Notify all watchers of cluster information changes
Optional.ofNullable(WATCHERS.remove(event.getGroup()))
.ifPresent(watchers -> watchers.parallelStream().forEach(this::notifyWatcher));
}
}
private void notifyWatcher(Watcher<HttpContext> watcher) {
- watcher.setDone(true);
- sendWatcherResponse(watcher, HttpResponseStatus.OK);
- }
+ HttpContext context = watcher.getAsyncContext();
+ boolean isHttp2 = context instanceof HttpContext && context.isHttp2();
+
+ if (!isHttp2) {
+ watcher.setDone(true);
+ }
+
+ boolean isFirstResponse = !HTTP2_HEADERS_SENT.getOrDefault(watcher, false);
+ sendWatcherResponse(watcher, HttpResponseStatus.OK, false, isFirstResponse);
+ if (isFirstResponse && isHttp2) {
+ HTTP2_HEADERS_SENT.put(watcher, true);
+ }
- private void sendWatcherResponse(Watcher<HttpContext> watcher, HttpResponseStatus nettyStatus) {
+ // Update watcher's term to the latest term to prevent infinite loop
+ // This ensures that when registryWatcher is called, it won't trigger notifyWatcher again
String group = watcher.getGroup();
+ Long latestTerm = GROUP_UPDATE_TERM.get(group);
+ if (latestTerm != null && latestTerm > watcher.getTerm()) {
+ watcher.setTerm(latestTerm);
+ }
+
+ // For HTTP/2, re-register the watcher to continue listening for future updates
+ if (isHttp2 && !watcher.isDone()) {
+ registryWatcher(watcher);
+ }
+ }
+ /**
+ * Send watcher response to the client.
+ *
+ * @param watcher the watcher instance
+ * @param nettyStatus the HTTP status code
+ * @param closeStream whether to close the HTTP/2 stream (endStream=true)
+ * @param sendHeaders whether to send HTTP/2 headers frame (only needed for first response)
+ */
+ private void sendWatcherResponse(
+ Watcher<HttpContext> watcher, HttpResponseStatus nettyStatus, boolean closeStream, boolean sendHeaders) {
+
HttpContext context = watcher.getAsyncContext();
if (!(context instanceof HttpContext)) {
logger.warn(
"Unsupported context type for watcher on group {}: {}",
- group,
+ watcher.getGroup(),
context != null ? context.getClass().getName() : "null");
return;
}
Review Comment:
The check `context instanceof HttpContext` is redundant and will always be true since the parameter is already declared as `HttpContext`. Additionally, the warning message at line 155 will never be reached in normal circumstances. Consider removing this check or adjusting the logic if this is meant to check for a more specific subtype.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org