diff --git a/app/assets/javascripts/editor/schema/ci.json b/app/assets/javascripts/editor/schema/ci.json
index 70f1dddcac9225f9cea23fcc510d01dea2ea79b9..1fb6f606b6b865398c14ee42bd56491efdd41513 100644
--- a/app/assets/javascripts/editor/schema/ci.json
+++ b/app/assets/javascripts/editor/schema/ci.json
@@ -774,6 +774,9 @@
               },
               "allow_failure": {
                 "$ref": "#/definitions/allow_failure"
+              },
+              "needs": {
+                "$ref": "#/definitions/rulesNeeds"
               }
             }
           },
@@ -936,6 +939,39 @@
       "markdownDescription": "Used in conjunction with 'when: delayed' to set how long to delay before starting a job. e.g. '5', 5 seconds, 30 minutes, 1 week, etc. [Learn More](https://docs.gitlab.com/ee/ci/jobs/job_control.html#run-a-job-after-a-delay).",
       "minLength": 1
     },
+    "rulesNeeds": {
+      "markdownDescription": "Use needs in rules to update job needs for specific conditions. When a condition matches a rule, the job's needs configuration is completely replaced with the needs in the rule. [Learn More](https://docs.gitlab.com/ee/ci/yaml/index.html#rulesneeds).",
+      "type": "array",
+      "items": {
+        "oneOf": [
+          {
+            "type": "string"
+          },
+          {
+            "type": "object",
+            "additionalProperties": false,
+            "properties": {
+              "job": {
+                "type": "string",
+                "minLength": 1,
+                "description": "Name of a job that is defined in the pipeline."
+              },
+              "artifacts": {
+                "type": "boolean",
+                "description": "Download artifacts of the job in needs."
+              },
+              "optional": {
+                "type": "boolean",
+                "description": "Whether the job needs to be present in the pipeline to run ahead of the current job."
+              }
+            },
+            "required": [
+              "job"
+            ]
+          }
+        ]
+      }
+    },
     "allow_failure": {
       "markdownDescription": "Allow job to fail. A failed job does not cause the pipeline to fail. [Learn More](https://docs.gitlab.com/ee/ci/yaml/#allow_failure).",
       "oneOf": [
diff --git a/config/feature_flags/development/introduce_rules_with_needs.yml b/config/feature_flags/development/introduce_rules_with_needs.yml
new file mode 100644
index 0000000000000000000000000000000000000000..8b2940438ee19965a22895fc49965e58a63c7aab
--- /dev/null
+++ b/config/feature_flags/development/introduce_rules_with_needs.yml
@@ -0,0 +1,8 @@
+---
+name: introduce_rules_with_needs
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/112725
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/394769
+milestone: '16.0'
+type: development
+group: group::pipeline authoring
+default_enabled: false
diff --git a/doc/ci/yaml/index.md b/doc/ci/yaml/index.md
index 8eb7e5a13dfd876ba050a3f41ac169f1e3e92100..49dbd08c90b3ac33b8abb483f5f8ab91b08b5cf9 100644
--- a/doc/ci/yaml/index.md
+++ b/doc/ci/yaml/index.md
@@ -3666,6 +3666,55 @@ If the rule matches, then the job is a manual job with `allow_failure: true`.
 - The rule-level `rules:allow_failure` overrides the job-level [`allow_failure`](#allow_failure),
   and only applies when the specific rule triggers the job.
 
+#### `rules:needs`
+
+> [Introduced](https://gitlab.com/gitlab-org/gitlab/-/issues/31581) in GitLab 16.0 [with a flag](../../user/feature_flags.md) named `introduce_rules_with_needs`. Disabled by default.
+
+Use `needs` in rules to update a job's [`needs`](#needs) for specific conditions. When a condition matches a rule, the job's `needs` configuration is completely replaced with the `needs` in the rule.
+
+**Keyword type**: Job-specific. You can use it only as part of a job.
+
+**Possible inputs**:
+
+- An array of job names as strings.
+- A hash with a job name, optionally with additional attributes.
+- An empty array (`[]`), to set the job needs to none when the specific condition is met.
+
+**Example of `rules:needs`**:
+
+```yaml
+build-dev:
+  stage: build
+  rules:
+    - if: $CI_COMMIT_BRANCH != $CI_DEFAULT_BRANCH
+  script: echo "Feature branch, so building dev version..."
+
+build-prod:
+  stage: build
+  rules:
+    - if: $CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH
+  script: echo "Default branch, so building prod version..."
+
+specs:
+  stage: test
+  needs: ['build-dev']
+  rules:
+    - if: $CI_COMMIT_REF_NAME == $CI_DEFAULT_BRANCH
+      needs: ['build-prod']
+    - when: on_success # Run the job in other cases
+  script: echo "Running dev specs by default, or prod specs when default branch..."
+```
+
+In this example:
+
+- If the pipeline runs on a branch that is not the default branch, the `specs` job needs the `build-dev` job (default behavior).
+- If the pipeline runs on the default branch, and therefore the rule matches the condition, the `specs` job needs the `build-prod` job instead.
+
+**Additional details**:
+
+- `needs` in rules override any `needs` defined at the job-level. When overridden, the behavior is same as [job-level `needs`](#needs).
+- `needs` in rules can accept [`artifacts`](#needsartifacts) and [`optional`](#needsoptional).
+
 #### `rules:variables`
 
 > - [Introduced](https://gitlab.com/gitlab-org/gitlab/-/issues/209864) in GitLab 13.7.
diff --git a/lib/gitlab/ci/build/rules.rb b/lib/gitlab/ci/build/rules.rb
index dee95534b0755416240e67e25d0b6660beb76959..bc7aad1b186042cf120395f36f0b5a3c99f7529d 100644
--- a/lib/gitlab/ci/build/rules.rb
+++ b/lib/gitlab/ci/build/rules.rb
@@ -6,12 +6,14 @@ module Build
       class Rules
         include ::Gitlab::Utils::StrongMemoize
 
-        Result = Struct.new(:when, :start_in, :allow_failure, :variables, :errors) do
+        Result = Struct.new(:when, :start_in, :allow_failure, :variables, :needs, :errors) do
           def build_attributes
             {
               when: self.when,
               options: { start_in: start_in }.compact,
-              allow_failure: allow_failure
+              allow_failure: allow_failure,
+              scheduling_type: (:dag if needs),
+              needs_attributes: needs&.[](:job)
             }.compact
           end
 
@@ -33,13 +35,14 @@ def evaluate(pipeline, context)
               matched_rule.attributes[:when] || @default_when,
               matched_rule.attributes[:start_in],
               matched_rule.attributes[:allow_failure],
-              matched_rule.attributes[:variables]
+              matched_rule.attributes[:variables],
+              (matched_rule.attributes[:needs] if Feature.enabled?(:introduce_rules_with_needs, pipeline.project))
             )
           else
             Result.new('never')
           end
         rescue Rule::Clause::ParseError => e
-          Result.new('never', nil, nil, nil, [e.message])
+          Result.new('never', nil, nil, nil, nil, [e.message])
         end
 
         private
diff --git a/lib/gitlab/ci/config/entry/rules/rule.rb b/lib/gitlab/ci/config/entry/rules/rule.rb
index 63bf1b38ac68c0599e86fd7aa32d56419b56c430..1e7f6056a651e90b4d1dcb85b2f28ad090cdcc85 100644
--- a/lib/gitlab/ci/config/entry/rules/rule.rb
+++ b/lib/gitlab/ci/config/entry/rules/rule.rb
@@ -9,7 +9,7 @@ class Rules::Rule < ::Gitlab::Config::Entry::Node
           include ::Gitlab::Config::Entry::Configurable
           include ::Gitlab::Config::Entry::Attributable
 
-          ALLOWED_KEYS = %i[if changes exists when start_in allow_failure variables].freeze
+          ALLOWED_KEYS = %i[if changes exists when start_in allow_failure variables needs].freeze
           ALLOWED_WHEN = %w[on_success on_failure always never manual delayed].freeze
 
           attributes :if, :exists, :when, :start_in, :allow_failure
@@ -20,6 +20,11 @@ class Rules::Rule < ::Gitlab::Config::Entry::Node
           entry :variables, Entry::Variables,
             description: 'Environment variables to define for rule conditions.'
 
+          entry :needs, Entry::Needs,
+            description: 'Needs configuration to define for rule conditions.',
+            metadata: { allowed_needs: %i[job] },
+            inherit: false
+
           validations do
             validates :config, presence: true
             validates :config, type: { with: Hash }
@@ -46,7 +51,8 @@ class Rules::Rule < ::Gitlab::Config::Entry::Node
           def value
             config.merge(
               changes: (changes_value if changes_defined?),
-              variables: (variables_value if variables_defined?)
+              variables: (variables_value if variables_defined?),
+              needs: (needs_value if needs_defined?)
             ).compact
           end
 
diff --git a/lib/gitlab/ci/yaml_processor.rb b/lib/gitlab/ci/yaml_processor.rb
index 63242d60c85d437ed79be34b3b601984f5955540..c69d9218a664b91b654f7d27a252fb6e9f0f28fe 100644
--- a/lib/gitlab/ci/yaml_processor.rb
+++ b/lib/gitlab/ci/yaml_processor.rb
@@ -94,16 +94,20 @@ def validate_dynamic_child_pipeline_dependencies!(name, job)
       end
 
       def validate_job_needs!(name, job)
-        return unless needs = job.dig(:needs, :job)
+        validate_needs_specification!(name, job.dig(:needs, :job))
 
-        validate_duplicate_needs!(name, needs)
+        job[:rules]&.each do |rule|
+          validate_needs_specification!(name, rule.dig(:needs, :job))
+        end
+      end
+
+      def validate_needs_specification!(name, needs)
+        return unless needs
 
         needs.each do |need|
           validate_job_dependency!(name, need[:name], 'need', optional: need[:optional])
         end
-      end
 
-      def validate_duplicate_needs!(name, needs)
         duplicated_needs =
           needs
           .group_by { |need| need[:name] }
diff --git a/spec/frontend/editor/schema/ci/ci_schema_spec.js b/spec/frontend/editor/schema/ci/ci_schema_spec.js
index 87208ec7aa8b431a116ace18fead6713141d62c6..51fcf26c39aa467872909d2f29bd00e2648158be 100644
--- a/spec/frontend/editor/schema/ci/ci_schema_spec.js
+++ b/spec/frontend/editor/schema/ci/ci_schema_spec.js
@@ -27,6 +27,7 @@ import CacheYaml from './yaml_tests/positive_tests/cache.yml';
 import FilterYaml from './yaml_tests/positive_tests/filter.yml';
 import IncludeYaml from './yaml_tests/positive_tests/include.yml';
 import RulesYaml from './yaml_tests/positive_tests/rules.yml';
+import RulesNeedsYaml from './yaml_tests/positive_tests/rules_needs.yml';
 import ProjectPathYaml from './yaml_tests/positive_tests/project_path.yml';
 import VariablesYaml from './yaml_tests/positive_tests/variables.yml';
 import JobWhenYaml from './yaml_tests/positive_tests/job_when.yml';
@@ -46,6 +47,7 @@ import ProjectPathIncludeLeadSlashYaml from './yaml_tests/negative_tests/project
 import ProjectPathIncludeNoSlashYaml from './yaml_tests/negative_tests/project_path/include/no_slash.yml';
 import ProjectPathIncludeTailSlashYaml from './yaml_tests/negative_tests/project_path/include/tailing_slash.yml';
 import RulesNegativeYaml from './yaml_tests/negative_tests/rules.yml';
+import RulesNeedsNegativeYaml from './yaml_tests/negative_tests/rules_needs.yml';
 import TriggerNegative from './yaml_tests/negative_tests/trigger.yml';
 import VariablesInvalidOptionsYaml from './yaml_tests/negative_tests/variables/invalid_options.yml';
 import VariablesInvalidSyntaxDescYaml from './yaml_tests/negative_tests/variables/invalid_syntax_desc.yml';
@@ -88,6 +90,7 @@ describe('positive tests', () => {
       JobWhenYaml,
       HooksYaml,
       RulesYaml,
+      RulesNeedsYaml,
       VariablesYaml,
       ProjectPathYaml,
       IdTokensYaml,
@@ -121,6 +124,7 @@ describe('negative tests', () => {
       IncludeNegativeYaml,
       JobWhenNegativeYaml,
       RulesNegativeYaml,
+      RulesNeedsNegativeYaml,
       TriggerNegative,
       VariablesInvalidOptionsYaml,
       VariablesInvalidSyntaxDescYaml,
diff --git a/spec/frontend/editor/schema/ci/yaml_tests/negative_tests/rules_needs.yml b/spec/frontend/editor/schema/ci/yaml_tests/negative_tests/rules_needs.yml
new file mode 100644
index 0000000000000000000000000000000000000000..f2f1eb118f879fbd35fd6101fc15e75955cc38a4
--- /dev/null
+++ b/spec/frontend/editor/schema/ci/yaml_tests/negative_tests/rules_needs.yml
@@ -0,0 +1,46 @@
+# invalid rules:needs
+lint_job:
+  script: exit 0
+  rules:
+    - if: $var == null
+      needs:
+
+# invalid rules:needs
+lint_job_2:
+  script: exit 0
+  rules:
+    - if: $var == null
+      needs: [20]
+
+# invalid rules:needs
+lint_job_3:
+  script: exit 0
+  rules:
+    - if: $var == null
+      needs: 
+        - job:      
+
+# invalid rules:needs
+lint_job_5:
+  script: exit 0
+  rules:
+    - if: $var == null
+      needs: 
+        - pipeline: 5
+
+# invalid rules:needs
+lint_job_6:
+  script: exit 0
+  rules:
+    - if: $var == null
+      needs: 
+        - project: namespace/group/project-name
+
+# invalid rules:needs
+lint_job_7:
+  script: exit 0
+  rules:
+    - if: $var == null
+      needs: 
+        - pipeline: 5
+          job: lint_job_6
diff --git a/spec/frontend/editor/schema/ci/yaml_tests/positive_tests/rules_needs.yml b/spec/frontend/editor/schema/ci/yaml_tests/positive_tests/rules_needs.yml
new file mode 100644
index 0000000000000000000000000000000000000000..a4a5183dcf4ec9a7298529f9fcdf4e2eecae4fda
--- /dev/null
+++ b/spec/frontend/editor/schema/ci/yaml_tests/positive_tests/rules_needs.yml
@@ -0,0 +1,32 @@
+# valid workflow:rules:needs
+pre_lint_job:
+  script: exit 0
+  rules:
+    - if: $var == null
+
+lint_job:
+  script: exit 0
+  rules:
+    - if: $var == null
+
+rspec_job:
+  script: exit 0
+  rules:
+    - if: $var == null
+      needs: [lint_job]
+
+job:
+  needs: [rspec_job]
+  script: exit 0
+  rules:
+    - if: $var == null
+      needs: 
+        - job: lint_job
+          artifacts: false
+          optional: true
+        - job: pre_lint_job
+          artifacts: true
+          optional: false
+        - rspec_job
+    - if: $var == true
+      needs: [lint_job, pre_lint_job]
\ No newline at end of file
diff --git a/spec/lib/gitlab/ci/build/rules_spec.rb b/spec/lib/gitlab/ci/build/rules_spec.rb
index e82dcd0254d24e65af1e7d91d4575c6665274d41..1ece0f6b7b9d0ea79e69597eda35ce40c8a1a36c 100644
--- a/spec/lib/gitlab/ci/build/rules_spec.rb
+++ b/spec/lib/gitlab/ci/build/rules_spec.rb
@@ -181,6 +181,108 @@
       end
     end
 
+    context 'with needs' do
+      context 'when single needs is specified' do
+        let(:rule_list) do
+          [{ if: '$VAR == null', needs: [{ name: 'test', artifacts: true, optional: false }] }]
+        end
+
+        it {
+          is_expected.to eq(described_class::Result.new('on_success', nil, nil, nil,
+            [{ name: 'test', artifacts: true, optional: false }], nil))
+        }
+      end
+
+      context 'when multiple needs are specified' do
+        let(:rule_list) do
+          [{ if: '$VAR == null',
+             needs: [{ name: 'test', artifacts: true, optional: false },
+               { name: 'rspec', artifacts: true, optional: false }] }]
+        end
+
+        it {
+          is_expected.to eq(described_class::Result.new('on_success', nil, nil, nil,
+            [{ name: 'test', artifacts: true, optional: false },
+              { name: 'rspec', artifacts: true, optional: false }], nil))
+        }
+      end
+
+      context 'when there are no needs specified' do
+        let(:rule_list) { [{ if: '$VAR == null' }] }
+
+        it { is_expected.to eq(described_class::Result.new('on_success', nil, nil, nil, nil, nil)) }
+      end
+
+      context 'when need is specified with additional attibutes' do
+        let(:rule_list) do
+          [{ if: '$VAR == null', needs: [{
+            artifacts: true,
+            name: 'test',
+            optional: false,
+            when: 'never'
+          }] }]
+        end
+
+        it {
+          is_expected.to eq(
+            described_class::Result.new('on_success', nil, nil, nil,
+              [{ artifacts: true, name: 'test', optional: false, when: 'never' }], nil))
+        }
+      end
+
+      context 'when feature flag is disabled' do
+        before do
+          stub_feature_flags(introduce_rules_with_needs: false)
+        end
+
+        context 'with needs' do
+          context 'when single needs is specified' do
+            let(:rule_list) do
+              [{ if: '$VAR == null', needs: [{ name: 'test', artifacts: true, optional: false }] }]
+            end
+
+            it {
+              is_expected.to eq(described_class::Result.new('on_success', nil, nil, nil, nil, nil))
+            }
+          end
+
+          context 'when multiple needs are specified' do
+            let(:rule_list) do
+              [{ if: '$VAR == null',
+                 needs: [{ name: 'test', artifacts: true, optional: false },
+                   { name: 'rspec', artifacts: true, optional: false }] }]
+            end
+
+            it {
+              is_expected.to eq(described_class::Result.new('on_success', nil, nil, nil, nil, nil))
+            }
+          end
+
+          context 'when there are no needs specified' do
+            let(:rule_list) { [{ if: '$VAR == null' }] }
+
+            it { is_expected.to eq(described_class::Result.new('on_success', nil, nil, nil, nil, nil)) }
+          end
+
+          context 'when need is specified with additional attibutes' do
+            let(:rule_list) do
+              [{ if: '$VAR == null', needs:  [{
+                artifacts: true,
+                name: 'test',
+                optional: false,
+                when: 'never'
+              }] }]
+            end
+
+            it {
+              is_expected.to eq(
+                described_class::Result.new('on_success', nil, nil, nil, nil, nil))
+            }
+          end
+        end
+      end
+    end
+
     context 'with variables' do
       context 'with matching rule' do
         let(:rule_list) { [{ if: '$VAR == null', variables: { MY_VAR: 'my var' } }] }
@@ -208,9 +310,10 @@
     let(:start_in) { nil }
     let(:allow_failure) { nil }
     let(:variables) { nil }
+    let(:needs) { nil }
 
     subject(:result) do
-      Gitlab::Ci::Build::Rules::Result.new(when_value, start_in, allow_failure, variables)
+      Gitlab::Ci::Build::Rules::Result.new(when_value, start_in, allow_failure, variables, needs)
     end
 
     describe '#build_attributes' do
@@ -221,6 +324,45 @@
       it 'compacts nil values' do
         is_expected.to eq(options: {}, when: 'on_success')
       end
+
+      context 'scheduling_type' do
+        context 'when rules have needs' do
+          context 'single need' do
+            let(:needs) do
+              { job: [{ name: 'test' }] }
+            end
+
+            it 'saves needs' do
+              expect(subject[:needs_attributes]).to eq([{ name: "test" }])
+            end
+
+            it 'adds schedule type to the build_attributes' do
+              expect(subject[:scheduling_type]).to eq(:dag)
+            end
+          end
+
+          context 'multiple needs' do
+            let(:needs) do
+              { job: [{ name: 'test' }, { name: 'test_2', artifacts: true, optional: false }] }
+            end
+
+            it 'saves needs' do
+              expect(subject[:needs_attributes]).to match_array([{ name: "test" },
+                { name: 'test_2', artifacts: true, optional: false }])
+            end
+
+            it 'adds schedule type to the build_attributes' do
+              expect(subject[:scheduling_type]).to eq(:dag)
+            end
+          end
+        end
+
+        context 'when rules do not have needs' do
+          it 'does not add schedule type to the build_attributes' do
+            expect(subject.key?(:scheduling_type)).to be_falsy
+          end
+        end
+      end
     end
 
     describe '#pass?' do
diff --git a/spec/lib/gitlab/ci/pipeline/seed/build_spec.rb b/spec/lib/gitlab/ci/pipeline/seed/build_spec.rb
index 86a111112832421415d74e67cff83ccbed1621b9..9d5a9bc8058b5d0b0e314477b2df269ff4ed4b09 100644
--- a/spec/lib/gitlab/ci/pipeline/seed/build_spec.rb
+++ b/spec/lib/gitlab/ci/pipeline/seed/build_spec.rb
@@ -109,6 +109,104 @@
       end
     end
 
+    context 'with job:rules:[needs:]' do
+      context 'with a single rule' do
+        let(:job_needs_attributes) { [{ name: 'rspec' }] }
+
+        context 'when job has needs set' do
+          context 'when rule evaluates to true' do
+            let(:attributes) do
+              { name: 'rspec',
+                ref: 'master',
+                needs_attributes: job_needs_attributes,
+                rules: [{ if: '$VAR == null', needs: { job: [{ name: 'build-job' }] } }] }
+            end
+
+            it 'overrides the job needs' do
+              expect(subject).to include(needs_attributes: [{ name: 'build-job' }])
+            end
+          end
+
+          context 'when rule evaluates to false' do
+            let(:attributes) do
+              { name: 'rspec',
+                ref: 'master',
+                needs_attributes: job_needs_attributes,
+                rules: [{ if: '$VAR == true', needs: { job: [{ name: 'build-job' }] } }] }
+            end
+
+            it 'keeps the job needs' do
+              expect(subject).to include(needs_attributes: job_needs_attributes)
+            end
+          end
+
+          context 'with subkeys: artifacts, optional' do
+            let(:attributes) do
+              { name: 'rspec',
+                ref: 'master',
+                rules:
+                [
+                  { if: '$VAR == null',
+                    needs: {
+                      job: [{
+                        name: 'build-job',
+                        optional: false,
+                        artifacts: true
+                      }]
+                    } }
+                ] }
+            end
+
+            context 'when rule evaluates to true' do
+              it 'sets the job needs as well as the job subkeys' do
+                expect(subject[:needs_attributes]).to match_array([{ name: 'build-job', optional: false, artifacts: true }])
+              end
+
+              it 'sets the scheduling type to dag' do
+                expect(subject[:scheduling_type]).to eq(:dag)
+              end
+            end
+          end
+        end
+
+        context 'with multiple rules' do
+          context 'when a rule evaluates to true' do
+            let(:attributes) do
+              { name: 'rspec',
+                ref: 'master',
+                needs_attributes: job_needs_attributes,
+                rules: [
+                  { if: '$VAR == true', needs: { job: [{ name: 'rspec-1' }] } },
+                  { if: '$VAR2 == true', needs: { job: [{ name: 'rspec-2' }] } },
+                  { if: '$VAR3 == null', needs: { job: [{ name: 'rspec' }, { name: 'lint' }] } }
+                ] }
+            end
+
+            it 'overrides the job needs' do
+              expect(subject).to include(needs_attributes: [{ name: 'rspec' }, { name: 'lint' }])
+            end
+          end
+
+          context 'when all rules evaluates to false' do
+            let(:attributes) do
+              { name: 'rspec',
+                ref: 'master',
+                needs_attributes: job_needs_attributes,
+                rules: [
+                  { if: '$VAR == true', needs: { job: [{ name: 'rspec-1' }] }  },
+                  { if: '$VAR2 == true', needs: { job: [{ name: 'rspec-2' }] } },
+                  { if: '$VAR3 == true', needs: { job: [{ name: 'rspec-3' }] } }
+                ] }
+            end
+
+            it 'keeps the job needs' do
+              expect(subject).to include(needs_attributes: job_needs_attributes)
+            end
+          end
+        end
+      end
+    end
+
     context 'with job:tags' do
       let(:attributes) do
         {
diff --git a/spec/lib/gitlab/ci/yaml_processor_spec.rb b/spec/lib/gitlab/ci/yaml_processor_spec.rb
index a35dd968cd6648e9392afb5d602ffc6da42ce948..f8c2889798fde65f96be15d7ff79ac0894e2cdf2 100644
--- a/spec/lib/gitlab/ci/yaml_processor_spec.rb
+++ b/spec/lib/gitlab/ci/yaml_processor_spec.rb
@@ -659,6 +659,191 @@ module Ci
 
             it_behaves_like 'has warnings and expected error', /build job: need test is not defined in current or prior stages/
           end
+
+          describe '#validate_job_needs!' do
+            context "when all validations pass" do
+              let(:config) do
+                <<-EOYML
+                    stages:
+                      - lint
+                    lint_job:
+                      needs: [lint_job_2]
+                      stage: lint
+                      script: 'echo lint_job'
+                      rules:
+                        - if: $var == null
+                          needs:
+                            - lint_job_2
+                            - job: lint_job_3
+                              optional: true
+                    lint_job_2:
+                      stage: lint
+                      script: 'echo job'
+                      rules:
+                        - if: $var == null
+                    lint_job_3:
+                      stage: lint
+                      script: 'echo job'
+                      rules:
+                        - if: $var == null
+                EOYML
+              end
+
+              it 'returns a valid response' do
+                expect(subject).to be_valid
+                expect(subject).to be_instance_of(Gitlab::Ci::YamlProcessor::Result)
+              end
+            end
+
+            context 'needs as array' do
+              context 'single need in following stage' do
+                let(:config) do
+                  <<-EOYML
+                      stages:
+                        - lint
+                        - test
+                      lint_job:
+                        stage: lint
+                        script: 'echo lint_job'
+                        rules:
+                          - if: $var == null
+                            needs: [test_job]
+                      test_job:
+                        stage: test
+                        script: 'echo job'
+                        rules:
+                          - if: $var == null
+                  EOYML
+                end
+
+                it_behaves_like 'returns errors', 'lint_job job: need test_job is not defined in current or prior stages'
+              end
+
+              context 'multiple needs in the following stage' do
+                let(:config) do
+                  <<-EOYML
+                      stages:
+                        - lint
+                        - test
+                      lint_job:
+                        stage: lint
+                        script: 'echo lint_job'
+                        rules:
+                          - if: $var == null
+                            needs: [test_job, test_job_2]
+                      test_job:
+                        stage: test
+                        script: 'echo job'
+                        rules:
+                          - if: $var == null
+                      test_job_2:
+                        stage: test
+                        script: 'echo job'
+                        rules:
+                          - if: $var == null
+                  EOYML
+                end
+
+                it_behaves_like 'returns errors', 'lint_job job: need test_job is not defined in current or prior stages'
+              end
+
+              context 'single need in following state - hyphen need' do
+                let(:config) do
+                  <<-EOYML
+                      stages:
+                        - lint
+                        - test
+                      lint_job:
+                        stage: lint
+                        script: 'echo lint_job'
+                        rules:
+                          - if: $var == null
+                            needs:
+                              - test_job
+                      test_job:
+                        stage: test
+                        script: 'echo job'
+                        rules:
+                          - if: $var == null
+                  EOYML
+                end
+
+                it_behaves_like 'returns errors', 'lint_job job: need test_job is not defined in current or prior stages'
+              end
+
+              context 'when there are duplicate needs (string and hash)' do
+                let(:config) do
+                  <<-EOYML
+                      stages:
+                        - test
+                      test_job_1:
+                        stage: test
+                        script: 'echo lint_job'
+                        rules:
+                          - if: $var == null
+                            needs:
+                              - test_job_2
+                              - job: test_job_2
+                      test_job_2:
+                        stage: test
+                        script: 'echo job'
+                        rules:
+                          - if: $var == null
+                  EOYML
+                end
+
+                it_behaves_like 'returns errors', 'test_job_1 has the following needs duplicated: test_job_2.'
+              end
+            end
+
+            context 'rule needs as hash' do
+              context 'single hash need in following stage' do
+                let(:config) do
+                  <<-EOYML
+                      stages:
+                        - lint
+                        - test
+                      lint_job:
+                        stage: lint
+                        script: 'echo lint_job'
+                        rules:
+                          - if: $var == null
+                            needs:
+                              - job: test_job
+                                artifacts: false
+                                optional: false
+                      test_job:
+                        stage: test
+                        script: 'echo job'
+                        rules:
+                          - if: $var == null
+                  EOYML
+                end
+
+                it_behaves_like 'returns errors', 'lint_job job: need test_job is not defined in current or prior stages'
+              end
+            end
+
+            context 'job rule need does not exist' do
+              let(:config) do
+                <<-EOYML
+                  build:
+                    stage: build
+                    script: echo
+                    rules:
+                      - when: always
+                  test:
+                    stage: test
+                    script: echo
+                    rules:
+                      - if: $var == null
+                        needs: [unknown_job]
+                EOYML
+              end
+
+              it_behaves_like 'has warnings and expected error', /test job: undefined need: unknown_job/
+            end
+          end
         end
       end
 
diff --git a/spec/services/ci/create_pipeline_service/rules_spec.rb b/spec/services/ci/create_pipeline_service/rules_spec.rb
index 19f9e7e3e4a68f2c3a30129597d618bf64a2ab90..87112137675de26a1a1a3c3037bc580110ef5f6a 100644
--- a/spec/services/ci/create_pipeline_service/rules_spec.rb
+++ b/spec/services/ci/create_pipeline_service/rules_spec.rb
@@ -386,6 +386,109 @@ def find_job(name)
           expect(regular_job.allow_failure).to eq(true)
         end
       end
+
+      context 'with needs:' do
+        let(:config) do
+          <<-EOY
+            job1:
+              script: ls
+
+            job2:
+              script: ls
+              rules:
+                - if: $var == null
+                  needs: [job1]
+                - when: on_success
+
+            job3:
+              script: ls
+              rules:
+                - if: $var == null
+                  needs: [job1]
+                - needs: [job2]
+
+            job4:
+              script: ls
+              needs: [job1]
+              rules:
+                - if: $var == null
+                  needs: [job2]
+                - when: on_success
+                  needs: [job3]
+          EOY
+        end
+
+        let(:job1) { pipeline.builds.find_by(name: 'job1') }
+        let(:job2) { pipeline.builds.find_by(name: 'job2') }
+        let(:job3) { pipeline.builds.find_by(name: 'job3') }
+        let(:job4) { pipeline.builds.find_by(name: 'job4') }
+
+        context 'when the `$var` rule matches' do
+          it 'creates a pipeline with overridden needs' do
+            expect(pipeline).to be_persisted
+            expect(build_names).to contain_exactly('job1', 'job2', 'job3', 'job4')
+
+            expect(job1.needs).to be_empty
+            expect(job2.needs).to contain_exactly(an_object_having_attributes(name: 'job1'))
+            expect(job3.needs).to contain_exactly(an_object_having_attributes(name: 'job1'))
+            expect(job4.needs).to contain_exactly(an_object_having_attributes(name: 'job2'))
+          end
+        end
+
+        context 'when the `$var` rule does not match' do
+          let(:initialization_params) { base_initialization_params.merge(variables_attributes: variables_attributes) }
+
+          let(:variables_attributes) do
+            [{ key: 'var', secret_value: 'SOME_VAR' }]
+          end
+
+          it 'creates a pipeline with overridden needs' do
+            expect(pipeline).to be_persisted
+            expect(build_names).to contain_exactly('job1', 'job2', 'job3', 'job4')
+
+            expect(job1.needs).to be_empty
+            expect(job2.needs).to be_empty
+            expect(job3.needs).to contain_exactly(an_object_having_attributes(name: 'job2'))
+            expect(job4.needs).to contain_exactly(an_object_having_attributes(name: 'job3'))
+          end
+        end
+
+        context 'when the FF introduce_rules_with_needs is disabled' do
+          before do
+            stub_feature_flags(introduce_rules_with_needs: false)
+          end
+
+          context 'when the `$var` rule matches' do
+            it 'creates a pipeline without overridden needs' do
+              expect(pipeline).to be_persisted
+              expect(build_names).to contain_exactly('job1', 'job2', 'job3', 'job4')
+
+              expect(job1.needs).to be_empty
+              expect(job2.needs).to be_empty
+              expect(job3.needs).to be_empty
+              expect(job4.needs).to contain_exactly(an_object_having_attributes(name: 'job1'))
+            end
+          end
+
+          context 'when the `$var` rule does not match' do
+            let(:initialization_params) { base_initialization_params.merge(variables_attributes: variables_attributes) }
+
+            let(:variables_attributes) do
+              [{ key: 'var', secret_value: 'SOME_VAR' }]
+            end
+
+            it 'creates a pipeline without overridden needs' do
+              expect(pipeline).to be_persisted
+              expect(build_names).to contain_exactly('job1', 'job2', 'job3', 'job4')
+
+              expect(job1.needs).to be_empty
+              expect(job2.needs).to be_empty
+              expect(job3.needs).to be_empty
+              expect(job4.needs).to contain_exactly(an_object_having_attributes(name: 'job1'))
+            end
+          end
+        end
+      end
     end
 
     context 'changes:' do