diff --git a/lib/stream/consumers.js b/lib/stream/consumers.js
index ffe6e531205e7f..4566eff6a30f7f 100644
--- a/lib/stream/consumers.js
+++ b/lib/stream/consumers.js
@@ -63,6 +63,9 @@ async function text(stream) {
     else
       str += dec.decode(chunk, { stream: true });
   }
+  // Flush the streaming TextDecoder so that any pending
+  // incomplete multibyte characters are handled.
+  str += dec.decode(undefined, { stream: false });
   return str;
 }
 
diff --git a/src/node_i18n.cc b/src/node_i18n.cc
index 8a08a95bf75d42..b1b3f5d1749a4f 100644
--- a/src/node_i18n.cc
+++ b/src/node_i18n.cc
@@ -442,11 +442,24 @@ void ConverterObject::Decode(const FunctionCallbackInfo<Value>& args) {
   UErrorCode status = U_ZERO_ERROR;
   MaybeStackBuffer<UChar> result;
   MaybeLocal<Object> ret;
-  size_t limit = converter->min_char_size() * input.length();
+
+  UBool flush = (flags & CONVERTER_FLAGS_FLUSH) == CONVERTER_FLAGS_FLUSH;
+
+  // When flushing the final chunk, the limit is the maximum
+  // of either the input buffer length or the number of pending
+  // characters times the min char size.
+  size_t limit = converter->min_char_size() *
+      (!flush ?
+          input.length() :
+          std::max(
+              input.length(),
+              static_cast<size_t>(
+                  ucnv_toUCountPending(converter->conv(), &status))));
+  status = U_ZERO_ERROR;
+
   if (limit > 0)
     result.AllocateSufficientStorage(limit);
 
-  UBool flush = (flags & CONVERTER_FLAGS_FLUSH) == CONVERTER_FLAGS_FLUSH;
   auto cleanup = OnScopeLeave([&]() {
     if (flush) {
       // Reset the converter state.
diff --git a/test/parallel/test-stream-consumers.js b/test/parallel/test-stream-consumers.js
index 8f6a9deb1c27dc..1a3b6ce00200d4 100644
--- a/test/parallel/test-stream-consumers.js
+++ b/test/parallel/test-stream-consumers.js
@@ -232,3 +232,17 @@ const kArrayBuffer =
   stream.write({});
   stream.end({});
 }
+
+{
+  const stream = new TransformStream();
+  text(stream.readable).then(common.mustCall((str) => {
+    // Incomplete utf8 character is flushed as a replacement char
+    assert.strictEqual(str.charCodeAt(0), 0xfffd);
+  }));
+  const writer = stream.writable.getWriter();
+  Promise.all([
+    writer.write(new Uint8Array([0xe2])),
+    writer.write(new Uint8Array([0x82])),
+    writer.close(),
+  ]).then(common.mustCall());
+}