Fix StreamObserver object calls onCompleted() without first calling onNext() will fail.

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerHttpChannelObserver.java

@@ -188,6 +188,9 @@ protected void customizeHeaders(HttpHeaders headers, Throwable throwable, HttpOu
     }
 
     protected final void sendMetadata(HttpMetadata metadata) {
+        if (headerSent) {
+            return;
+        }
         getHttpChannel().writeHeader(metadata);
         headerSent = true;
         if (LOGGER.isDebugEnabled()) {
@@ -327,6 +330,10 @@ protected String getContentType() {
         return responseEncoder.contentType();
     }
 
+    protected boolean isHeaderSent() {
+        return headerSent;
+    }
+
     protected void customizeTrailers(HttpHeaders headers, Throwable throwable) {
         List<BiConsumer<HttpHeaders, Throwable>> trailersCustomizers = this.trailersCustomizers;
         if (trailersCustomizers != null) {

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http1/Http1SseServerChannelObserver.java

@@ -35,6 +35,14 @@ public void setResponseEncoder(HttpMessageEncoder responseEncoder) {
         super.setResponseEncoder(new ServerSentEventEncoder(responseEncoder));
     }
 
+    @Override
+    protected void doOnCompleted(Throwable throwable) {
+        if (!isHeaderSent()) {
+            sendMetadata(encodeHttpMetadata(true));
+        }
+        super.doOnCompleted(throwable);
+    }
+
     @Override
     protected HttpMetadata encodeHttpMetadata(boolean endStream) {
         return super.encodeHttpMetadata(endStream)

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/Http2SseServerChannelObserver.java

@@ -40,4 +40,13 @@ protected HttpMetadata encodeHttpMetadata(boolean endStream) {
         return super.encodeHttpMetadata(endStream)
                 .header(HttpHeaderNames.CACHE_CONTROL.getKey(), HttpConstants.NO_CACHE);
     }
+
+    @Override
+    protected void doOnCompleted(Throwable throwable) {
+        // if throwable is not null, the header will be flushed by super.doOnCompleted(throwable)
+        if (!isHeaderSent() && throwable == null) {
+            sendMetadata(encodeHttpMetadata(true));
+        }
+        super.doOnCompleted(throwable);
+    }
 }