[java-dfa] IDEA-300181 Report when the same stream is consumed more than once.

Co-authored-by: Tagir Valeev <tagir.valeev@jetbrains.com>

GitOrigin-RevId: 929f69d8e22285cae8906761df6b39ec46ce985c

java/java-analysis-api/resources/messages/JavaAnalysisBundle.properties

@@ -85,6 +85,8 @@ dataflow.message.constant.expression=Result of <code>#ref</code> #loc is always
 dataflow.message.constant.value=Value <code>#ref</code> #loc is always ''{0}''
 dataflow.method.fails.with.null.argument=Method will throw an exception when parameter is null
 dataflow.message.unknown.nullability=\ (unknown nullability)
+dataflow.message.stream.consumed.always=Stream has already been linked or consumed
+dataflow.message.stream.consumed=Stream might have already been linked or consumed
 dataflow.not.precise={0} is complex: data flow results could be imprecise
 dataflow.too.complex={0} is too complex to analyze by data flow algorithm
 
@@ -488,6 +490,7 @@ special.field.collection.size=Size
 special.field.unboxed.value=Unboxed value
 special.field.optional.value=Optional value
 special.field.enum.ordinal=Enum ordinal
+special.field.consumed.stream=Linked or consumed stream
 text.unused.import.in.template=Unused import (specified in template)
 text.raw.ctor.reference.with.type.parameters=Raw constructor reference with explicit type parameters for constructor
 

java/java-analysis-impl/src/com/intellij/codeInspection/dataFlow/ConsumedStreamUtils.java

@@ -0,0 +1,47 @@
+// Copyright 2000-2022 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
+package com.intellij.codeInspection.dataFlow;
+
+import com.intellij.psi.PsiMethod;
+import com.siyeh.ig.callMatcher.CallMatcher;
+
+import static com.intellij.psi.CommonClassNames.JAVA_UTIL_STREAM_BASE_STREAM;
+import static com.siyeh.ig.callMatcher.CallMatcher.anyOf;
+import static com.siyeh.ig.callMatcher.CallMatcher.instanceCall;
+
+public final class ConsumedStreamUtils {
+
+  private static final CallMatcher ON_CLOSE_STREAM = instanceCall(JAVA_UTIL_STREAM_BASE_STREAM, "onClose");
+
+  private static final CallMatcher CLOSE_STREAM_MATCHERS = instanceCall(JAVA_UTIL_STREAM_BASE_STREAM, "close");
+
+  private static final CallMatcher SKIP_STREAM = instanceCall(JAVA_UTIL_STREAM_BASE_STREAM, "parallel", "sequential");
+  private static final CallMatcher MARK_AND_CONSUMED_STREAM =
+    instanceCall(JAVA_UTIL_STREAM_BASE_STREAM, "iterator", "spliterator",
+                 "forEach", "forEachOrdered", "toArray", "reduce", "collect", "sum", "min", "max", "count", "average", "summaryStatistics",
+                 "anyMatch", "allMatch", "nonMatch", "findFirst", "findAny", "toList", "filter", "map", "mapToObj", "mapToInt", "mapToLong",
+                 "mapToDouble", "flatMap", "flatMapToInt", "flatMapToLong", "flatMapToDouble", "mapMulti", "mapMultiToInt",
+                 "mapMultiToLong", "mapMultiToDouble", "distinct", "sorted", "peek", "limit", "skip", "takeWhile", "dropWhile",
+                 "asLongStream", "asDoubleStream", "boxed");
+  private static final CallMatcher CALL_TO_MARK_CONSUMED =
+    anyOf(MARK_AND_CONSUMED_STREAM, CLOSE_STREAM_MATCHERS);
+  private static final CallMatcher NONE_LEAKS_STREAM =
+    anyOf(MARK_AND_CONSUMED_STREAM, CLOSE_STREAM_MATCHERS, ON_CLOSE_STREAM, SKIP_STREAM);
+
+  private static final CallMatcher IS_CHECKED_CALL = anyOf(MARK_AND_CONSUMED_STREAM, ON_CLOSE_STREAM);
+
+  public static boolean isCheckedCallForConsumedStream(PsiMethod method) {
+    return IS_CHECKED_CALL.methodMatches(method);
+  }
+
+  public static CallMatcher getSkipStreamMatchers() {
+    return SKIP_STREAM;
+  }
+
+  public static CallMatcher getCallToMarkConsumedStreamMatchers() {
+    return CALL_TO_MARK_CONSUMED;
+  }
+
+  public static CallMatcher getAllNonLeakStreamMatchers() {
+    return NONE_LEAKS_STREAM;
+  }
+}

java/java-analysis-impl/src/com/intellij/codeInspection/dataFlow/DataFlowInspectionBase.java

@@ -267,6 +267,7 @@ public abstract class DataFlowInspectionBase extends AbstractBaseJavaLocalInspec
 
     reportDuplicateAssignments(reporter, visitor);
     reportPointlessSameArguments(reporter, visitor);
+    reportStreamConsumed(holder, visitor);
   }
 
   private static void reportRedundantInstanceOf(DataFlowInstructionVisitor visitor, ProblemReporter reporter) {
@@ -567,6 +568,14 @@ public abstract class DataFlowInspectionBase extends AbstractBaseJavaLocalInspec
     });
   }
 
+  private void reportStreamConsumed(ProblemsHolder holder, DataFlowInstructionVisitor visitor) {
+    visitor.streamConsumed().forKeyValue((psiElement, alwaysFails) -> {
+      if (!REPORT_UNSOUND_WARNINGS && !alwaysFails) return;
+      holder.registerProblem(psiElement, JavaAnalysisBundle.message(alwaysFails ? "dataflow.message.stream.consumed.always" :
+                                                                    "dataflow.message.stream.consumed"));
+    });
+  }
+
   private static void reportArrayStoreProblems(ProblemsHolder holder, DataFlowInstructionVisitor visitor) {
     visitor.getArrayStoreProblems().forEach(
       (assignment, types) -> holder.registerProblem(assignment.getOperationSign(), JavaAnalysisBundle

java/java-analysis-impl/src/com/intellij/codeInspection/dataFlow/DataFlowInstructionVisitor.java

@@ -55,6 +55,7 @@ final class DataFlowInstructionVisitor implements JavaDfaListener {
   private final Map<PsiAssignmentExpression, Pair<PsiType, PsiType>> myArrayStoreProblems = new HashMap<>();
   private final Map<PsiArrayAccessExpression, ThreeState> myOutOfBoundsArrayAccesses = new HashMap<>();
   private final Map<PsiExpression, ThreeState> myNegativeArraySizes = new HashMap<>();
+  private final Map<PsiElement, StateInfo> myStreamConsumed = new HashMap<>();
   private final Set<PsiElement> myReceiverMutabilityViolation = new HashSet<>();
   private final Set<PsiElement> myArgumentMutabilityViolation = new HashSet<>();
   private final Map<PsiExpression, Boolean> mySameValueAssigned = new HashMap<>();
@@ -197,6 +198,11 @@ final class DataFlowInstructionVisitor implements JavaDfaListener {
     return StreamEx.ofKeys(myNegativeArraySizes, ThreeState.YES::equals);
   }
 
+  EntryStream<PsiElement, Boolean> streamConsumed() {
+    return EntryStream.of(myStreamConsumed).filterValues(StateInfo::shouldReport).mapToValue(
+      (element, info) -> info.alwaysFails());
+  }
+
   StreamEx<PsiCallExpression> alwaysFailingCalls() {
     return StreamEx.ofKeys(myFailingCalls, v -> v).map(ContractFailureProblem::getAnchor).select(PsiCallExpression.class).distinct();
   }
@@ -365,6 +371,10 @@ final class DataFlowInstructionVisitor implements JavaDfaListener {
       StateInfo info = myStateInfos.computeIfAbsent(nullabilityProblem, k -> new StateInfo());
       info.update(state, ok);
     }
+    else if (problem instanceof ConsumedStreamProblem consumedStreamProblem) {
+      myStreamConsumed.computeIfAbsent(consumedStreamProblem.getAnchor(), e -> new StateInfo())
+        .update(state, ThreeState.fromBoolean(failed != ThreeState.YES));
+    }
   }
 
   @Override

java/java-analysis-impl/src/com/intellij/codeInspection/dataFlow/HardcodedContracts.java

@@ -15,6 +15,7 @@ import com.siyeh.ig.callMatcher.CallMatcher;
 import com.siyeh.ig.psiutils.ConstructionUtils;
 import com.siyeh.ig.psiutils.MethodUtils;
 import com.siyeh.ig.psiutils.TypeUtils;
+import org.jetbrains.annotations.Contract;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
@@ -69,6 +70,19 @@ public final class HardcodedContracts {
     return NO_PARAMETER_LEAK_METHODS.methodMatches(method);
   }
 
+  private static final CallMatcher NO_QUALIFIER_LEAK_MATCHERS =
+    //not fully clear, but they don't affect on CONSUMED_STREAM and other tracked parameters
+    ConsumedStreamUtils.getAllNonLeakStreamMatchers();
+
+  /**
+   * @param method   method to test
+   * @return true if given method doesn't spoil its qualifier
+   */
+  @Contract(value = "null -> false", pure = true)
+  public static boolean isKnownNoQualifierLeak(@Nullable PsiMethod method) {
+    return NO_QUALIFIER_LEAK_MATCHERS.methodMatches(method);
+  }
+
   @FunctionalInterface
   interface ContractProvider {
     List<MethodContract> getContracts(PsiMethodCallExpression call, int paramCount);
@@ -184,7 +198,9 @@ public final class HardcodedContracts {
       instanceCall("java.time.YearMonth", "isBefore", "isAfter")
     ), ContractProvider.of(
       singleConditionContract(ContractValue.qualifier(), RelationType.EQ, ContractValue.argument(0), returnFalse())
-    ));
+    ))
+    //for propagation CONSUMED_STREAM
+    .register(ConsumedStreamUtils.getSkipStreamMatchers(), ContractProvider.of(trivialContract(returnThis())));
 
   private static @NotNull ContractProvider getArraycopyContract() {
     ContractValue src = ContractValue.argument(0);

java/java-analysis-impl/src/com/intellij/codeInspection/dataFlow/SideEffectHandlers.java

@@ -6,6 +6,7 @@ import com.intellij.codeInspection.dataFlow.memory.DfaMemoryState;
 import com.intellij.codeInspection.dataFlow.rangeSet.LongRangeBinOp;
 import com.intellij.codeInspection.dataFlow.rangeSet.LongRangeSet;
 import com.intellij.codeInspection.dataFlow.types.DfIntType;
+import com.intellij.codeInspection.dataFlow.types.DfStreamStateType;
 import com.intellij.codeInspection.dataFlow.types.DfType;
 import com.intellij.codeInspection.dataFlow.types.DfTypes;
 import com.intellij.codeInspection.dataFlow.value.DfaValue;
@@ -20,6 +21,7 @@ import java.util.Objects;
 import java.util.Set;
 
 import static com.intellij.codeInspection.dataFlow.jvm.SpecialField.COLLECTION_SIZE;
+import static com.intellij.codeInspection.dataFlow.jvm.SpecialField.CONSUMED_STREAM;
 import static com.intellij.psi.CommonClassNames.*;
 import static com.intellij.util.ObjectUtils.tryCast;
 import static com.siyeh.ig.callMatcher.CallMatcher.anyOf;
@@ -62,7 +64,9 @@ class SideEffectHandlers {
                     instanceCall(JAVA_UTIL_MAP, "remove").parameterTypes(JAVA_LANG_OBJECT, JAVA_LANG_OBJECT)),
               (factory, state, arguments) -> collectionRemove(factory, state, arguments, false))
     .register(instanceCall(JAVA_UTIL_LIST, "remove").parameterTypes("int"),
-              (factory, state, arguments) -> collectionRemove(factory, state, arguments, true));
+              (factory, state, arguments) -> collectionRemove(factory, state, arguments, true))
+    .register(ConsumedStreamUtils.getCallToMarkConsumedStreamMatchers(),
+              (factory, state, arguments) -> streamConsume(factory, state, arguments));
 
   private static void collectionAdd(DfaValueFactory factory, DfaMemoryState state, DfaCallArguments arguments, boolean list) {
     DfaVariableValue size = tryCast(COLLECTION_SIZE.createValue(factory, arguments.myQualifier), DfaVariableValue.class);
@@ -79,7 +83,7 @@ class SideEffectHandlers {
           resultSize = COLLECTION_SIZE.getDefaultValue();
         }
       }
-      updateSize(state, size, resultSize);
+      updateState(state, size, resultSize);
     }
   }
 
@@ -92,7 +96,7 @@ class SideEffectHandlers {
         resultSize = sizeType.eval(DfTypes.intRange(LongRangeSet.range(strict ? 1 : 0, 1)), LongRangeBinOp.MINUS)
           .meet(DfTypes.intRange(JvmPsiRangeSetUtil.indexRange()));
       }
-      updateSize(state, size, resultSize);
+      updateState(state, size, resultSize);
     }
   }
 
@@ -120,7 +124,7 @@ class SideEffectHandlers {
           resultSize = COLLECTION_SIZE.getDefaultValue();
         }
       }
-      updateSize(state, size, resultSize);
+      updateState(state, size, resultSize);
     }
   }
 
@@ -133,14 +137,21 @@ class SideEffectHandlers {
         LongRangeSet newSize = sizeType.getRange().fromRelation(RelationType.LE).meet(JvmPsiRangeSetUtil.indexRange());
         resultSize = sizeType.join(DfTypes.intRange(newSize));
       }
-      updateSize(state, size, resultSize);
+      updateState(state, size, resultSize);
     }
   }
 
   private static void collectionClear(DfaValueFactory factory, DfaMemoryState state, DfaCallArguments arguments) {
     DfaVariableValue size = tryCast(COLLECTION_SIZE.createValue(factory, arguments.myQualifier), DfaVariableValue.class);
     if (size != null) {
-      updateSize(state, size, DfTypes.intValue(0));
+      updateState(state, size, DfTypes.intValue(0));
+    }
+  }
+
+  private static void streamConsume(DfaValueFactory factory, DfaMemoryState state, DfaCallArguments arguments) {
+    DfaVariableValue consumedStream = tryCast(CONSUMED_STREAM.createValue(factory, arguments.myQualifier), DfaVariableValue.class);
+    if (consumedStream != null) {
+      updateState(state, consumedStream, DfStreamStateType.CONSUMED);
     }
   }
 
@@ -163,7 +174,7 @@ class SideEffectHandlers {
     void handleSideEffect(DfaValueFactory factory, DfaMemoryState state, DfaCallArguments arguments);
   }
 
-  private static void updateSize(DfaMemoryState state, DfaVariableValue var, DfType type) {
+  private static void updateState(DfaMemoryState state, DfaVariableValue var, DfType type) {
     // Dependent states may appear which we are not tracking (e.g. one visible list is sublist of another list)
     // so let's conservatively flush everything that could be affected
     state.flushFieldsQualifiedBy(Set.of(Objects.requireNonNull(var.getQualifier())));

java/java-analysis-impl/src/com/intellij/codeInspection/dataFlow/java/ControlFlowAnalyzer.java

@@ -16,6 +16,7 @@ import com.intellij.codeInspection.dataFlow.jvm.SpecialField;
 import com.intellij.codeInspection.dataFlow.jvm.TrapTracker;
 import com.intellij.codeInspection.dataFlow.jvm.descriptors.*;
 import com.intellij.codeInspection.dataFlow.jvm.problems.ArrayIndexProblem;
+import com.intellij.codeInspection.dataFlow.jvm.problems.ConsumedStreamProblem;
 import com.intellij.codeInspection.dataFlow.jvm.problems.ContractFailureProblem;
 import com.intellij.codeInspection.dataFlow.jvm.problems.NegativeArraySizeProblem;
 import com.intellij.codeInspection.dataFlow.jvm.transfer.*;
@@ -28,6 +29,7 @@ import com.intellij.codeInspection.dataFlow.lang.ir.ControlFlow.ControlFlowOffse
 import com.intellij.codeInspection.dataFlow.lang.ir.ControlFlow.DeferredOffset;
 import com.intellij.codeInspection.dataFlow.rangeSet.LongRangeBinOp;
 import com.intellij.codeInspection.dataFlow.rangeSet.LongRangeSet;
+import com.intellij.codeInspection.dataFlow.types.DfStreamStateType;
 import com.intellij.codeInspection.dataFlow.types.DfType;
 import com.intellij.codeInspection.dataFlow.types.DfTypes;
 import com.intellij.codeInspection.dataFlow.value.*;
@@ -835,6 +837,9 @@ public class ControlFlowAnalyzer extends JavaElementVisitor {
             generateBoxingUnboxingInstructionFor(returnValue, LambdaUtil.getFunctionalInterfaceReturnType(lambdaExpression));
           }
         }
+        if (InheritanceUtil.isInheritor(returnValue.getType(), JAVA_UTIL_STREAM_BASE_STREAM)) {
+          addConsumedStreamCheckInstructions(returnValue, null);
+        }
         addInstruction(new PopInstruction());
       }
 
@@ -843,6 +848,17 @@ public class ControlFlowAnalyzer extends JavaElementVisitor {
     finishElement(statement);
   }
 
+  private void addConsumedStreamCheckInstructions(@Nullable PsiExpression reference, @Nullable String exception) {
+    if (reference == null) {
+      return;
+    }
+    DfaControlTransferValue transferValue = exception != null ? createTransfer(exception) : null;
+    addInstruction(new DupInstruction());
+    addInstruction(new UnwrapDerivedVariableInstruction(SpecialField.CONSUMED_STREAM));
+    addInstruction(new EnsureInstruction(new ConsumedStreamProblem(reference), RelationType.NE, DfStreamStateType.CONSUMED, transferValue));
+    addInstruction(new PopInstruction());
+  }
+
   @Override
   public void visitSwitchLabelStatement(@NotNull PsiSwitchLabelStatement statement) {
     startElement(statement);
@@ -1891,7 +1907,12 @@ public class ControlFlowAnalyzer extends JavaElementVisitor {
     PsiExpression[] expressions = call.getArgumentList().getExpressions();
     PsiReferenceExpression methodExpression = call.getMethodExpression();
     JavaResolveResult result = methodExpression.advancedResolve(false);
-    PsiParameter[] parameters = result.getElement() instanceof PsiMethod method ? method.getParameterList().getParameters() : null;
+    PsiMethod method = ObjectUtils.tryCast(result.getElement(), PsiMethod.class);
+    PsiParameter[] parameters = method != null ? method.getParameterList().getParameters() : null;
+
+    if (method != null && ConsumedStreamUtils.isCheckedCallForConsumedStream(method)) {
+      addConsumedStreamCheckInstructions(methodExpression.getQualifierExpression(), "java.lang.IllegalStateException");
+    }
 
     for (int i = 0; i < expressions.length; i++) {
       PsiExpression paramExpr = expressions[i];

java/java-analysis-impl/src/com/intellij/codeInspection/dataFlow/java/JavaDfaValueFactory.java

@@ -32,6 +32,7 @@ import org.jetbrains.annotations.Nullable;
 
 import java.util.List;
 
+import static com.intellij.psi.CommonClassNames.JAVA_UTIL_STREAM_BASE_STREAM;
 import static com.intellij.util.ObjectUtils.tryCast;
 
 /**
@@ -202,6 +203,8 @@ public class JavaDfaValueFactory {
     }
     if (target instanceof PsiMethod) {
       PsiMethod method = (PsiMethod)target;
+      // Assume that methods returning stream always return a new one
+      if (InheritanceUtil.isInheritor(method.getReturnType(), JAVA_UTIL_STREAM_BASE_STREAM)) return null;
       if (method.getParameterList().isEmpty() &&
           (PropertyUtilBase.isSimplePropertyGetter(method) || JavaMethodContractUtil.isPure(method) || isClassAnnotatedImmutable(method)) &&
           isContractAllowedForGetter(method)) {

java/java-analysis-impl/src/com/intellij/codeInspection/dataFlow/java/inliner/StreamChainInliner.java

@@ -23,7 +23,9 @@ import com.intellij.codeInspection.dataFlow.Mutability;
 import com.intellij.codeInspection.dataFlow.java.CFGBuilder;
 import com.intellij.codeInspection.dataFlow.jvm.JvmPsiRangeSetUtil;
 import com.intellij.codeInspection.dataFlow.jvm.SpecialField;
+import com.intellij.codeInspection.dataFlow.jvm.problems.ConsumedStreamProblem;
 import com.intellij.codeInspection.dataFlow.rangeSet.LongRangeSet;
+import com.intellij.codeInspection.dataFlow.types.DfStreamStateType;
 import com.intellij.codeInspection.dataFlow.types.DfType;
 import com.intellij.codeInspection.dataFlow.types.DfTypes;
 import com.intellij.codeInspection.dataFlow.value.DfaVariableValue;
@@ -565,7 +567,7 @@ public class StreamChainInliner implements CallInliner {
     void iteration(CFGBuilder builder) {
       if (myStreamSource != null) {
         builder.assignTo(myParameter).pop();
-        buildStreamCFG(builder, myChain, myStreamSource);
+        buildStreamCFG(builder, myChain, myStreamSource, false);
       } else {
         PsiExpression arg = myCall.getArgumentList().getExpressions()[0];
         PsiType outType = StreamApiUtil.getStreamElementType(myCall.getType());
@@ -830,11 +832,16 @@ public class StreamChainInliner implements CallInliner {
     }
     PsiExpression originalQualifier = firstStep.myCall.getMethodExpression().getQualifierExpression();
     if (originalQualifier == null) return false;
-    builder.pushUnknown()
-           .ifConditionIs(true)
-           .chain(b -> buildStreamCFG(b, firstStep, originalQualifier))
-           .end()
-           .push(DfTypes.typedObject(call.getType(), Nullability.NOT_NULL), call);
+    builder.pushExpression(originalQualifier)
+      .chain(b -> checkAndMarkConsumed(b, originalQualifier))
+      .pop()
+      .pushUnknown()
+      .ifConditionIs(true)
+      .chain(b -> buildStreamCFG(b, firstStep, originalQualifier, true))
+      .end()
+      .push(builder.getFactory().fromDfType(SpecialField.CONSUMED_STREAM.asDfType(DfStreamStateType.OPEN)
+                                                 .meet(DfTypes.LOCAL_OBJECT)
+                                                 .meet(DfTypes.NOT_NULL_OBJECT)));
     return true;
   }
 
@@ -844,12 +851,13 @@ public class StreamChainInliner implements CallInliner {
     Step firstStep = buildChain(qualifierCall, terminalStep);
     PsiExpression originalQualifier = firstStep.myCall.getMethodExpression().getQualifierExpression();
     if (originalQualifier == null) return false;
-    buildStreamCFG(builder, firstStep, originalQualifier);
+    buildStreamCFG(builder, firstStep, originalQualifier, false);
     firstStep.pushResult(builder);
     return true;
   }
 
-  private static void buildStreamCFG(CFGBuilder builder, Step firstStep, PsiExpression originalQualifier) {
+  private static void buildStreamCFG(CFGBuilder builder, Step firstStep, PsiExpression originalQualifier,
+                                     boolean originalQualifierAlreadyChecked) {
     PsiType inType = StreamApiUtil.getStreamElementType(originalQualifier.getType(), false);
     PsiMethodCallExpression sourceCall = tryCast(PsiUtil.skipParenthesizedExprDown(originalQualifier), PsiMethodCallExpression.class);
     if(STREAM_GENERATE.test(sourceCall)) {
@@ -918,9 +926,13 @@ public class StreamChainInliner implements CallInliner {
         .push(DfTypes.intValue(0))
         .ifCondition(RelationType.GT);
     } else {
+      if (!originalQualifierAlreadyChecked) {
+        builder
+          .pushExpression(originalQualifier)
+          .chain(b -> checkAndMarkConsumed(b, originalQualifier))
+          .pop();
+      }
       builder
-        .pushExpression(originalQualifier)
-        .pop()
         .chain(firstStep::before)
         .pushUnknown()
         .ifConditionIs(true);
@@ -930,6 +942,16 @@ public class StreamChainInliner implements CallInliner {
       .end();
   }
 
+  private static void checkAndMarkConsumed(CFGBuilder builder, PsiExpression qualifier) {
+    builder
+      .dup()
+      .unwrap(SpecialField.CONSUMED_STREAM)
+      .ensure(RelationType.NE, DfStreamStateType.CONSUMED, new ConsumedStreamProblem(qualifier), "java.lang.IllegalStateException")
+      .push(DfStreamStateType.CONSUMED)
+      .assign()
+      .pop();
+  }
+
   private static void makeMainLoop(CFGBuilder builder, Step firstStep, PsiType inType) {
     builder.doWhileUnknown()
            .assign(builder.createTempVariable(inType), DfTypes.typedObject(inType, DfaPsiUtil.getTypeNullability(inType)))

java/java-analysis-impl/src/com/intellij/codeInspection/dataFlow/java/inst/MethodCallInstruction.java

@@ -19,10 +19,12 @@ import com.intellij.codeInspection.dataFlow.memory.DfaMemoryState;
 import com.intellij.codeInspection.dataFlow.rangeSet.LongRangeSet;
 import com.intellij.codeInspection.dataFlow.types.DfConstantType;
 import com.intellij.codeInspection.dataFlow.types.DfReferenceType;
+import com.intellij.codeInspection.dataFlow.types.DfStreamStateType;
 import com.intellij.codeInspection.dataFlow.types.DfType;
 import com.intellij.codeInspection.dataFlow.value.*;
 import com.intellij.openapi.diagnostic.Logger;
 import com.intellij.psi.*;
+import com.intellij.psi.util.InheritanceUtil;
 import com.intellij.psi.util.PsiUtil;
 import com.intellij.psi.util.TypeConversionUtil;
 import com.intellij.util.ThreeState;
@@ -32,9 +34,14 @@ import com.siyeh.ig.psiutils.MethodUtils;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
-import java.util.*;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
 
+import static com.intellij.codeInspection.dataFlow.jvm.SpecialField.CONSUMED_STREAM;
 import static com.intellij.codeInspection.dataFlow.types.DfTypes.*;
+import static com.intellij.psi.CommonClassNames.JAVA_UTIL_STREAM_BASE_STREAM;
 import static com.intellij.util.ObjectUtils.tryCast;
 
 
@@ -396,6 +403,13 @@ public class MethodCallInstruction extends ExpressionPushingInstruction {
           !TypeConstraint.fromDfType(dfType).isComparedByEquals()) {
         dfType = dfType.meet(LOCAL_OBJECT);
       }
+      if (InheritanceUtil.isInheritor(type, JAVA_UTIL_STREAM_BASE_STREAM)) {
+        dfType = dfType.meet(((DfReferenceType)CONSUMED_STREAM.asDfType(DfStreamStateType.OPEN)).dropNullability());
+        if (myReturnNullability == Nullability.NOT_NULL) {
+          dfType = dfType.meet(LOCAL_OBJECT);
+        }
+      }
+
       return factory.fromDfType(dfType.meet(mutable.asDfType()));
     }
     LongRangeSet range = JvmPsiRangeSetUtil.typeRange(type, true);
@@ -447,7 +461,8 @@ public class MethodCallInstruction extends ExpressionPushingInstruction {
       }
     }
     TypeConstraint constraint = TypeConstraint.fromDfType(dfType);
-    if (!constraint.isArray() && (constraint.isComparedByEquals() || mayLeakThis(memState, argValues))) {
+    if (!HardcodedContracts.isKnownNoQualifierLeak(getTargetMethod()) &&
+        !constraint.isArray() && (constraint.isComparedByEquals() || mayLeakThis(memState, argValues))) {
       value = JavaDfaHelpers.dropLocality(value, memState);
     }
     return value;

java/java-analysis-impl/src/com/intellij/codeInspection/dataFlow/jvm/SpecialField.java

@@ -5,6 +5,7 @@ import com.intellij.codeInsight.Nullability;
 import com.intellij.codeInspection.dataFlow.*;
 import com.intellij.codeInspection.dataFlow.rangeSet.LongRangeSet;
 import com.intellij.codeInspection.dataFlow.types.DfReferenceType;
+import com.intellij.codeInspection.dataFlow.types.DfStreamStateType;
 import com.intellij.codeInspection.dataFlow.types.DfType;
 import com.intellij.codeInspection.dataFlow.types.DfTypes;
 import com.intellij.codeInspection.dataFlow.value.*;
@@ -276,6 +277,23 @@ public enum SpecialField implements DerivedVariableDescriptor {
     boolean isMyAccessor(PsiMember accessor) {
       return accessor instanceof PsiMethod && ENUM_ORDINAL_METHOD.methodMatches((PsiMethod)accessor);
     }
+  },
+  CONSUMED_STREAM("linkedOrConsumed", "special.field.consumed.stream", false) {
+    @Override
+    boolean isMyQualifierType(DfType type) {
+      TypeConstraint constraint = TypeConstraint.fromDfType(type);
+      return constraint.isSubtypeOf(JAVA_UTIL_STREAM_BASE_STREAM);
+    }
+
+    @Override
+    boolean isMyAccessor(PsiMember accessor) {
+      return false;
+    }
+
+    @Override
+    public @NotNull DfType getDefaultValue() {
+      return DfStreamStateType.UNKNOWN;
+    }
   };
 
   private static final SpecialField[] VALUES = values();

java/java-analysis-impl/src/com/intellij/codeInspection/dataFlow/jvm/problems/ConsumedStreamProblem.java

@@ -0,0 +1,11 @@
+// Copyright 2000-2022 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
+package com.intellij.codeInspection.dataFlow.jvm.problems;
+
+import com.intellij.psi.PsiElement;
+import org.jetbrains.annotations.NotNull;
+
+public class ConsumedStreamProblem extends JvmDfaProblem<PsiElement> {
+  public ConsumedStreamProblem(@NotNull PsiElement anchor) {
+    super(anchor);
+  }
+}

java/java-analysis-impl/src/com/intellij/codeInspection/dataFlow/types/DfStreamStateType.java

@@ -0,0 +1,70 @@
+// Copyright 2000-2022 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
+package com.intellij.codeInspection.dataFlow.types;
+
+import com.intellij.codeInspection.dataFlow.value.RelationType;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Represents linkedOrConsumed field for Stream
+ * UNKNOWN-->OPEN
+ * \     \
+ * ------->CONSUMED
+ */
+public enum DfStreamStateType implements DfType {
+  UNKNOWN, OPEN, CONSUMED;
+
+  @Override
+  public boolean isSuperType(@NotNull DfType other) {
+    if (other.equals(this) || other == DfType.BOTTOM) return true;
+    if (other instanceof DfStreamStateType && this == UNKNOWN) return true;
+    return false;
+  }
+
+  @Override
+  public @NotNull DfType join(@NotNull DfType other) {
+    if (other.equals(this) || other == DfType.BOTTOM) return this;
+    if (other instanceof DfStreamStateType) return UNKNOWN;
+    return DfType.TOP;
+  }
+
+  @Override
+  public DfType tryJoinExactly(@NotNull DfType other) {
+    if (other.isSuperType(this)) return other;
+    if (this.isSuperType(other)) return this;
+    return null;
+  }
+
+  @Override
+  public @NotNull DfType meet(@NotNull DfType other) {
+    if (other.isSuperType(this)) return this;
+    if (this.isSuperType(other)) return other;
+    return BOTTOM;
+  }
+
+  @Override
+  public @NotNull DfType fromRelation(@NotNull RelationType relationType) {
+    DfType afterRelation = switch (relationType) {
+      case EQ -> this;
+      case NE -> tryNegate();
+      default -> TOP;
+    };
+    return afterRelation == null ? TOP : afterRelation;
+  }
+
+  @Nullable
+  @Override
+  public DfType tryNegate() {
+    return switch (this) {
+      case UNKNOWN -> null;
+      case OPEN -> CONSUMED;
+      case CONSUMED -> OPEN;
+    };
+  }
+
+  @Override
+  public DfType correctTypeOnFlush(DfType typeBeforeFlush) {
+    if (typeBeforeFlush == CONSUMED) return typeBeforeFlush;
+    return this;
+  }
+}

java/java-tests/testData/inspection/dataFlow/fixture/ConsumedStream.java

@@ -0,0 +1,242 @@
+import java.util.Iterator;
+import java.util.Random;
+import java.util.stream.Stream;
+import org.jetbrains.annotations.Nullable;
+
+public class ConsumedStream {
+  public static void test1() {
+    Stream<String> stream = Stream.of("x");
+    stream.forEach(System.out::println);
+    <warning descr="Stream has already been linked or consumed">stream</warning>.forEach(System.out::println);
+  }
+
+  public static void test2() {
+    Stream<String> stream = Stream.of("x");
+    Stream<String> stream2 = stream.filter(String::isEmpty);
+    Stream<String> stream3 = <warning descr="Stream has already been linked or consumed">stream</warning>.filter(x -> !x.isEmpty());
+  }
+
+  public static void test3() {
+    Stream<String> stream = Stream.of("x");
+    for (int i = 0; i < 10; i++) {
+      <warning descr="Stream might have already been linked or consumed">stream</warning>.forEach(System.out::println);
+    }
+  }
+
+  public static void test4() {
+    Stream<String> stream = Stream.of("x");
+    stream.forEach(System.out::println);
+    if (Math.random() > 0.5) {
+      <warning descr="Stream has already been linked or consumed">stream</warning>.forEach(System.out::println);
+    }
+  }
+
+  public static void test5() {
+    Stream<String> stream = Stream.of("x");
+    switch (new Random().nextInt(10)) {
+      case 0:
+        stream.forEach(System.out::println);
+      case 1:
+        <warning descr="Stream might have already been linked or consumed">stream</warning>.forEach(System.out::println);
+    }
+  }
+
+  public static void test6() {
+    Stream<String> stream = Stream.of("x");
+    try {
+      method1();
+      stream.forEach(System.out::println);
+    } finally {
+      <warning descr="Stream might have already been linked or consumed">stream</warning>.forEach(System.out::println);
+    }
+  }
+
+  private static void method1() {}
+
+  public static void test7() {
+    Stream<String> stream = Stream.of("x");
+    try {
+      stream.forEach(System.out::println);
+    } catch (Exception e) {
+      <warning descr="Stream has already been linked or consumed">stream</warning>.forEach(System.out::println);
+    }
+  }
+
+  public static void test1E() {
+    Stream<String> stream = Stream.of("x");
+    int x = new Random().nextInt(10);
+    if (x < 2) {
+      stream.forEach(System.out::println);
+    }
+    if (x < 1) {
+      <warning descr="Stream has already been linked or consumed">stream</warning>.forEach(System.out::println);
+    }
+  }
+
+  public static void test2E() {
+    Stream<String> stream = Stream.of("x");
+    Stream<String> stream2 = stream.filter(String::isEmpty);
+    double random = Math.random();
+    Stream<String> stream3 = <warning descr="Stream has already been linked or consumed">stream</warning>.filter(x -> !x.isEmpty());
+  }
+
+  public static void test3E() {
+    Stream<String> stream = Stream.of("x");
+    double random = Math.random();
+
+    if (random > 0.5) {
+      stream.forEach(System.out::println);
+    }
+
+    if (random > 0.9) {
+      <warning descr="Stream has already been linked or consumed">stream</warning>.forEach(System.out::println);
+    }
+  }
+
+  public static void test6E() {
+    Stream<String> stream = Stream.of("x");
+    double random2 = Math.random();
+    Stream<String> stringStream = stream.filter(t -> true);
+    boolean parallel = stream.isParallel();
+    double random3 = Math.random();
+    method(stream);
+    <warning descr="Stream has already been linked or consumed">stream</warning>.filter(t -> true);
+  }
+
+  public static void test7E() {
+    Stream<String> stream = Stream.of("x");
+    Stream<String> parallel = stream.parallel().filter(t->true);
+    <warning descr="Stream has already been linked or consumed">stream</warning>.forEach(System.out::println);
+  }
+
+  public static void test8E() {
+    Stream<String> stream = Stream.of("x");
+    stream.forEach(System.out::println);
+    Stream<String> parallel = stream.parallel();
+    <warning descr="Stream has already been linked or consumed">parallel</warning>.forEach(System.out::println);
+  }
+
+  private static void method(Stream<String> stream) {
+    Iterator<String> iterator = stream.iterator();
+  }
+
+  public static Stream<String> test9E() {
+    Stream<String> stringStream = Stream.of("1");
+    Stream<String> stringStream1 = stringStream.filter(t -> true);
+    return <warning descr="Stream has already been linked or consumed">stringStream</warning>;
+  }
+
+  public static Stream<String> test10E() {
+    Stream<String> stringStream = Stream.of("1");
+    Stream<String> stringStream1 = stringStream.filter(t -> true);
+    return <warning descr="Stream has already been linked or consumed">stringStream.sequential()</warning>;
+  }
+
+  public static void test11E() {
+    Stream<Integer> integerStream = test11Em();
+    Stream<Integer> integerStream1 = integerStream.<warning descr="Method invocation 'filter' may produce 'NullPointerException'">filter</warning>(t -> true);
+    Stream<Integer> integerStream2 = <warning descr="Stream has already been linked or consumed">integerStream</warning>.filter(t -> true);
+  }
+
+  private static void test12E(Stream<Object> stream) {
+    stream.forEach(System.out::println);
+    String text = "noError";
+    try {
+      <warning descr="Stream has already been linked or consumed">stream</warning>.forEach(System.out::println);
+    } catch (IllegalStateException e) {
+      text = "IllegalStateException";
+    } catch (RuntimeException e) {
+      text = "RuntimeException";
+    }
+    if (<warning descr="Condition 'text.equals(\"IllegalStateException\")' is always 'true'">text.equals("IllegalStateException")</warning>) {
+      System.out.println("error");
+    }
+  }
+
+  private static void test13E(int x) {
+    Stream<String> stream = Stream.of("x");
+    if (x == 0) {
+      stream.forEach(System.out::println);
+    }
+    <warning descr="Stream might have already been linked or consumed">stream</warning>.forEach(System.out::println);
+  }
+
+  @Nullable
+  public static Stream<Integer> test11Em() {
+    return null;
+  }
+
+  public static void test1N() {
+    Stream<String> stream = Stream.of("x");
+    if (Math.random() > 0.5) {
+      stream.forEach(System.out::println);
+    } else {
+      stream.distinct().forEach(System.out::println);
+    }
+  }
+
+  public static void test2N() {
+    Stream<String> stream = Stream.of("x");
+    stream = stream.filter(x -> !x.isEmpty());
+    stream.forEach(System.out::println);
+  }
+
+  public static void test3N() {
+    Stream<String> stream = Stream.of("x");
+    System.out.println(stream.isParallel());
+    stream.forEach(System.out::println);
+  }
+
+  public static void test4N() {
+    Stream<String> stream = Stream.of("x");
+    switch (new Random().nextInt(10)) {
+      case 0:
+        stream.forEach(System.out::println);
+        break;
+      case 1:
+        stream.forEachOrdered(System.out::println);
+    }
+  }
+
+  public static void test5N() {
+    Stream<String> stream = Stream.of("x");
+    for (int i = 0; i < 10; i++) {
+      if (Math.random() > 0.5) {
+        stream.forEach(System.out::println);
+        break;
+      }
+    }
+  }
+
+  public static void test1NE() {
+    Stream<String> stream = Stream.of("x");
+    boolean b = Math.random() > 0.5;
+    if (b) {
+      stream.forEach(System.out::println);
+    }
+    if (!b) {
+      stream.forEach(System.out::println);
+    }
+  }
+
+  public static void test2NE() {
+    Stream<String> stream = Stream.of("x");
+    int x = new Random().nextInt(10);
+    if (x < 2) {
+      stream.forEach(System.out::println);
+    }
+    if (x > 5) {
+      stream.forEach(System.out::println);
+    }
+  }
+
+  public static Stream<String> test3NE() {
+    Stream<String> stringStream = Stream.of("1");
+    return stringStream;
+  }
+
+  public static void test4NE() {
+    Stream<String> stringStream = Stream.of("1");
+    stringStream.filter(t -> true);
+  }
+}

java/java-tests/testData/inspection/dataFlow/fixture/ConsumedStreamDifferentMethods.java

@@ -0,0 +1,99 @@
+import java.util.*;
+import java.util.stream.*;
+
+public class ConsumedStreamDifferentMethods {
+  public static void test1() {
+    Stream<String> stream = Stream.of("x");
+    Stream<String> stringStream = stream.filter(t -> true);
+    double random = Math.random();
+    Stream<String> stringStream1 = <warning descr="Stream has already been linked or consumed">stream</warning>.map(t -> t);
+  }
+
+  public static void test2() {
+    Stream<String> stream = Stream.empty();
+    IntStream intStream = stream.mapToInt(t -> 1);
+    double random = Math.random();
+    LongStream longStream = <warning descr="Stream has already been linked or consumed">stream</warning>.mapToLong(t -> 1L);
+  }
+
+  public static void test3() {
+    Stream<String> stream = Stream.generate(()->"x");
+    IntStream intStream = stream.mapToInt(t -> 1);
+    double random = Math.random();
+    LongStream longStream = <warning descr="Stream has already been linked or consumed">stream</warning>.mapToLong(t -> 1L);
+  }
+
+  public static void test4() {
+    Stream<String> stream = Stream.of("x");
+    Stream<String> stream2 = stream.peek(t->{});
+    double random = Math.random();
+    Stream<String> stream3 = <warning descr="Stream has already been linked or consumed">stream</warning>.skip(1);
+  }
+
+  public static void test5() {
+    IntStream stream = IntStream.of(1);
+    IntStream stream2 = stream.filter(t -> true);
+    double random = Math.random();
+    OptionalInt min = <warning descr="Stream has already been linked or consumed">stream</warning>.min();
+  }
+
+  public static void test6() {
+    IntStream stream = IntStream.empty();
+    IntStream stream2 = stream.map(t -> 2);
+    double random = Math.random();
+    IntSummaryStatistics intSummaryStatistics = <warning descr="Stream has already been linked or consumed">stream</warning>.summaryStatistics();
+  }
+
+  public static void test7() {
+    LongStream stream = LongStream.empty();
+    IntStream intStream = stream.mapToInt(t -> 1);
+    double random = Math.random();
+    OptionalLong max = <warning descr="Stream has already been linked or consumed">stream</warning>.max();
+  }
+
+  public static void test8() {
+    DoubleStream stream = DoubleStream.concat(DoubleStream.of(1), DoubleStream.of(2));
+    Stream<String> stream2 = stream.mapToObj(String::valueOf);
+    double random = Math.random();
+    OptionalDouble average = <warning descr="Stream has already been linked or consumed">stream</warning>.average();
+  }
+
+  public static void test9() {
+    ArrayList<String> strings = new ArrayList<>();
+    strings.add("x");
+    Stream<String> stream = strings.stream();
+    long count = stream.count();
+    double random = Math.random();
+    Stream<String> stringStream = <warning descr="Stream has already been linked or consumed">stream</warning>.onClose(() -> {
+    });
+  }
+
+  public static void test10() {
+    int[] ints = new int[3];
+    IntStream stream = Arrays.stream(ints);
+    stream.close();
+    double random = Math.random();
+    IntStream distinct = <warning descr="Stream has already been linked or consumed">stream</warning>.distinct();
+  }
+
+  public static void test11() {
+    ArrayList<String> strings = new ArrayList<>();
+    strings.add("x");
+    Spliterator<String> spliterator = strings.spliterator();
+    Stream<String> stream = StreamSupport.stream(spliterator, false);
+    List<String> collect = stream.collect(Collectors.toList());
+    double random = Math.random();
+    boolean parallel = stream.isParallel();
+    Optional<String> any = <warning descr="Stream has already been linked or consumed">stream</warning>.findAny();
+  }
+
+  public static void test12() {
+    ArrayList<String> strings = new ArrayList<>();
+    strings.add("x");
+    Stream<String> stream = strings.stream().filter(t->true);
+    List<String> collect = stream.collect(Collectors.toList());
+    double random = Math.random();
+    boolean parallel = stream.isParallel();
+    Optional<String> any = <warning descr="Stream has already been linked or consumed">stream</warning>.findAny();
+  }
+}

java/java-tests/testData/inspection/dataFlow/fixture/ConsumedStreamWithoutInline.java

@@ -0,0 +1,209 @@
+import java.util.Iterator;
+import java.util.List;
+import java.util.Random;
+import java.util.Spliterator;
+import java.util.stream.Stream;
+
+public class ConsumedStreamWithoutInline {
+  public static void test1() {
+    Stream<String> stream = Stream.of("x");
+    Iterator<String> iterator = stream.iterator();
+    Iterator<String> iterator2 = <warning descr="Stream has already been linked or consumed">stream</warning>.iterator();
+  }
+
+  public static void test2() {
+    Stream<String> stream = Stream.of("x");
+    Spliterator<String> spliterator = stream.spliterator();
+    Iterator<String> iterator = <warning descr="Stream has already been linked or consumed">stream</warning>.iterator();
+  }
+
+  public static void test3() {
+    Stream<String> stream = Stream.of("x");
+    for (int i = 0; i < 10; i++) {
+      Iterator<String> iterator = <warning descr="Stream might have already been linked or consumed">stream</warning>.iterator();
+    }
+  }
+
+  public static void test4() {
+    Stream<String> stream = Stream.of("x");
+    stream.forEach(System.out::println);
+    if (Math.random() > 0.5) {
+      Iterator<String> iterator = <warning descr="Stream has already been linked or consumed">stream</warning>.iterator();
+    }
+  }
+
+  public static void test5() {
+    Stream<String> stream = Stream.of("x");
+    switch (new Random().nextInt(10)) {
+      case 0:
+        Iterator<String> iterator = stream.iterator();
+      case 1:
+        Spliterator<String> spliterator = <warning descr="Stream might have already been linked or consumed">stream</warning>.spliterator();
+    }
+  }
+
+  public static void test6() {
+    Stream<String> stream = Stream.of("x");
+    try {
+      method1();
+      Iterator<String> iterator = stream.iterator();
+    } finally {
+      Spliterator<String> spliterator = <warning descr="Stream might have already been linked or consumed">stream</warning>.spliterator();
+    }
+  }
+
+  private static void method1() {}
+
+  public static void test7() {
+    Stream<String> stream = Stream.of("x");
+    try {
+      Iterator<String> iterator = stream.iterator();
+    } catch (Exception e) {
+      Spliterator<String> spliterator = <warning descr="Stream has already been linked or consumed">stream</warning>.spliterator();
+    }
+  }
+
+  public static void test1E() {
+    Stream<String> stream = Stream.of("x");
+    int x = new Random().nextInt(10);
+    if (x < 2) {
+      Iterator<String> iterator = stream.iterator();
+    }
+    if (x < 1) {
+      Spliterator<String> spliterator = <warning descr="Stream has already been linked or consumed">stream</warning>.spliterator();
+    }
+  }
+
+  public static void test2E() {
+    Stream<String> stream = Stream.of("x");
+    Iterator<String> iterator = stream.iterator();
+    double random = Math.random();
+    Iterator<String> iterator2 = <warning descr="Stream has already been linked or consumed">stream</warning>.iterator();
+  }
+
+  public static void test3E() {
+    Stream<String> stream = Stream.of("x");
+    double random = Math.random();
+
+    if (random > 0.5) {
+      Iterator<String> iterator = stream.iterator();
+    }
+
+    if (random > 0.9) {
+      Iterator<String> iterator2 = <warning descr="Stream has already been linked or consumed">stream</warning>.iterator();
+    }
+  }
+
+  public static void test6E() {
+    Stream<String> stream = Stream.of("x");
+    double random2 = Math.random();
+    Iterator<String> iterator = stream.iterator();
+    boolean parallel = stream.isParallel();
+    double random3 = Math.random();
+    method(stream);
+    Iterator<String> iterator2 = <warning descr="Stream has already been linked or consumed">stream</warning>.iterator();
+  }
+
+  private static void method(Stream<String> stream) {
+    Iterator<String> iterator = stream.iterator();
+  }
+
+  public static void test8E() {
+    Stream<String> stream = Stream.of("x");
+    Iterator<String> iterator = stream.iterator();
+    Stream<String> parallel = stream.parallel();
+    Iterator<String> iterator2 = <warning descr="Stream has already been linked or consumed">parallel</warning>.iterator();
+  }
+
+  public static Stream<String> test9E() {
+    Stream<String> stringStream = Stream.of("1");
+    Iterator<String> iterator1 = stringStream.iterator();
+    return <warning descr="Stream has already been linked or consumed">stringStream</warning>;
+  }
+
+  private static void test10E(Stream<Object> stream) {
+    Iterator<Object> iterator = stream.iterator();
+    String text = "noError";
+    try {
+      Iterator<Object> iterator2 = <warning descr="Stream has already been linked or consumed">stream</warning>.iterator();
+    } catch (IllegalStateException e) {
+      text = "IllegalStateException";
+    } catch (RuntimeException e) {
+      text = "RuntimeException";
+    }
+    if (<warning descr="Condition 'text.equals(\"IllegalStateException\")' is always 'true'">text.equals("IllegalStateException")</warning>) {
+      System.out.println("error");
+    }
+  }
+
+  private static void test11E(int x) {
+    Stream<String> stream = Stream.of("x");
+    if (x == 0) {
+      Iterator<String> iterator = stream.iterator();
+    }
+    Iterator<String> iterator2 = <warning descr="Stream might have already been linked or consumed">stream</warning>.iterator();
+  }
+
+  public static void test1N() {
+    Stream<String> stream = Stream.of("x");
+    if (Math.random() > 0.5) {
+      Iterator<String> iterator = stream.iterator();
+    } else {
+      Spliterator<String> spliterator = stream.spliterator();
+    }
+  }
+
+  public static void test3N() {
+    Stream<String> stream = Stream.of("x");
+    System.out.println(stream.isParallel());
+    Spliterator<String> spliterator = stream.spliterator();
+  }
+
+  public static void test4N() {
+    Stream<String> stream = Stream.of("x");
+    switch (new Random().nextInt(10)) {
+      case 0:
+        Iterator<String> iterator = stream.iterator();
+        break;
+      case 1:
+        Spliterator<String> spliterator = stream.spliterator();
+    }
+  }
+
+  public static void test5N() {
+    Stream<String> stream = Stream.of("x");
+    for (int i = 0; i < 10; i++) {
+      if (Math.random() > 0.5) {
+        Iterator<String> iterator = stream.iterator();
+        break;
+      }
+    }
+  }
+
+  public static void test1NE() {
+    Stream<String> stream = Stream.of("x");
+    boolean b = Math.random() > 0.5;
+    if (b) {
+      Spliterator<String> spliterator = stream.spliterator();
+    }
+    if (!b) {
+      Iterator<String> iterator = stream.iterator();
+    }
+  }
+
+  public static void test2NE() {
+    Stream<String> stream = Stream.of("x");
+    int x = new Random().nextInt(10);
+    if (x < 2) {
+      Spliterator<String> spliterator = stream.spliterator();
+    }
+    if (x > 5) {
+      Iterator<String> iterator = stream.iterator();
+    }
+  }
+
+  public void test3NE(List<String> list) {
+    Iterator<String> iterator1 = list.stream().iterator();
+    Iterator<String> iterator2 = list.stream().iterator();
+  }
+}

java/java-tests/testSrc/com/intellij/java/codeInspection/DataFlowInspection8Test.java

@@ -386,4 +386,7 @@ public class DataFlowInspection8Test extends DataFlowInspectionTestCase {
   public void testConstructorMethodReferenceNullability() { doTest(); }
   public void testCustomStreamImplementation() { doTest(); }
   public void testEmptyCollection() { doTest(); }
+  public void testConsumedStream() { doTest(); }
+  public void testConsumedStreamDifferentMethods() { doTest(); }
+  public void testConsumedStreamWithoutInline()  { doTest(); }
 }

java/java-tests/testSrc/com/intellij/java/codeInspection/dataFlow/types/DfStreamStateTypeTest.java

@@ -0,0 +1,65 @@
+// Copyright 2000-2022 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
+package com.intellij.java.codeInspection.dataFlow.types;
+
+import com.intellij.codeInspection.dataFlow.value.RelationType;
+import org.junit.Test;
+
+import static com.intellij.codeInspection.dataFlow.types.DfStreamStateType.*;
+import static org.junit.Assert.*;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+public class DfStreamStateTypeTest {
+
+  @Test
+  public void isSuperType() {
+    assertTrue(UNKNOWN.isSuperType(CONSUMED));
+    assertTrue(UNKNOWN.isSuperType(OPEN));
+    assertTrue(OPEN.isSuperType(OPEN));
+    assertFalse(OPEN.isSuperType(CONSUMED));
+    assertFalse(OPEN.isSuperType(UNKNOWN));
+    assertFalse(CONSUMED.isSuperType(UNKNOWN));
+    assertFalse(CONSUMED.isSuperType(OPEN));
+  }
+
+  @Test
+  public void join() {
+    assertEquals(UNKNOWN, UNKNOWN.join(CONSUMED));
+    assertEquals(OPEN, OPEN.join(OPEN));
+    assertEquals(CONSUMED, CONSUMED.join(CONSUMED));
+    assertEquals(UNKNOWN, OPEN.join(CONSUMED));
+    assertEquals(UNKNOWN, OPEN.join(UNKNOWN));
+  }
+
+  @Test
+  public void tryJoinExactly() {
+    assertEquals(UNKNOWN, UNKNOWN.tryJoinExactly(CONSUMED));
+    assertEquals(OPEN, OPEN.tryJoinExactly(OPEN));
+    assertEquals(CONSUMED, CONSUMED.tryJoinExactly(CONSUMED));
+    assertNull(OPEN.tryJoinExactly(CONSUMED));
+    assertEquals(UNKNOWN, OPEN.tryJoinExactly(UNKNOWN));
+  }
+
+  @Test
+  public void meet() {
+    assertEquals(CONSUMED, UNKNOWN.meet(CONSUMED));
+    assertEquals(OPEN, OPEN.meet(OPEN));
+    assertEquals(CONSUMED, CONSUMED.meet(CONSUMED));
+    assertEquals(BOTTOM, OPEN.meet(CONSUMED));
+    assertEquals(OPEN, OPEN.meet(UNKNOWN));
+  }
+
+  @Test
+  public void meetRelation() {
+    assertEquals(CONSUMED, UNKNOWN.meetRelation(RelationType.EQ, CONSUMED));
+    assertEquals(OPEN, OPEN.meetRelation(RelationType.EQ, OPEN));
+    assertEquals(CONSUMED, CONSUMED.meetRelation(RelationType.EQ, CONSUMED));
+    assertEquals(BOTTOM, OPEN.meetRelation(RelationType.EQ, CONSUMED));
+    assertEquals(OPEN, OPEN.meetRelation(RelationType.EQ, UNKNOWN));
+
+    assertEquals(OPEN, UNKNOWN.meetRelation(RelationType.NE, CONSUMED));
+    assertEquals(BOTTOM, OPEN.meetRelation(RelationType.NE, OPEN));
+    assertEquals(BOTTOM, CONSUMED.meetRelation(RelationType.NE, CONSUMED));
+    assertEquals(OPEN, OPEN.meetRelation(RelationType.NE, CONSUMED));
+    assertEquals(OPEN, OPEN.meetRelation(RelationType.NE, UNKNOWN));
+  }
+}

java/jdkAnnotations/java/util/stream/annotations.xml

@@ -5,6 +5,11 @@
     <item name='java.util.stream.BaseStream S parallel()'>
         <annotation name='org.jetbrains.annotations.NotNull'/>
     </item>
+    <item name='java.util.stream.BaseStream boolean isParallel()'>
+      <annotation name='org.jetbrains.annotations.Contract'>
+        <val name="pure" val="true"/>
+      </annotation>
+    </item>
     <item name='java.util.stream.BaseStream S sequential()'>
         <annotation name='org.jetbrains.annotations.NotNull'/>
     </item>