diff --git a/core/src/main/java/org/apache/rocketmq/streams/core/rstream/RStream.java b/core/src/main/java/org/apache/rocketmq/streams/core/rstream/RStream.java
index 7ff181be..85a6fb73 100644
--- a/core/src/main/java/org/apache/rocketmq/streams/core/rstream/RStream.java
+++ b/core/src/main/java/org/apache/rocketmq/streams/core/rstream/RStream.java
@@ -27,7 +27,7 @@ public interface RStream<T> {
 
     <O> RStream<O> map(ValueMapperAction<T, O> mapperAction);
 
-    <VR> RStream<T> flatMap(final ValueMapperAction<T, ? extends Iterable<? extends VR>> mapper);
+    <VR> RStream<VR> flatMap(final ValueMapperAction<T, ? extends Iterable<? extends VR>> mapper);
 
     RStream<T> filter(FilterAction<T> predictor);
 
diff --git a/core/src/main/java/org/apache/rocketmq/streams/core/rstream/RStreamImpl.java b/core/src/main/java/org/apache/rocketmq/streams/core/rstream/RStreamImpl.java
index f9da6ee7..d84626f1 100644
--- a/core/src/main/java/org/apache/rocketmq/streams/core/rstream/RStreamImpl.java
+++ b/core/src/main/java/org/apache/rocketmq/streams/core/rstream/RStreamImpl.java
@@ -75,7 +75,7 @@ public <O> RStream<O> map(ValueMapperAction<T, O> mapperAction) {
     }
 
     @Override
-    public <VR> RStream<T> flatMap(ValueMapperAction<T, ? extends Iterable<? extends VR>> mapper) {
+    public <VR> RStream<VR> flatMap(ValueMapperAction<T, ? extends Iterable<? extends VR>> mapper) {
         String name = OperatorNameMaker.makeName(FLAT_MAP_PREFIX, pipeline.getJobId());
 
         MultiValueChangeSupplier<T, VR> changeSupplier = new MultiValueChangeSupplier<>(mapper);