def METHOD_NAME(self, consumer):
    # Full integration test to ensure that when a subscription receives an update
    # the `QuerySubscriptionConsumer` successfully retries the subscription and
    # calls the correct callback, which should result in an incident being created.
    message = {
        "version": 3,
        "payload": {
            "subscription_id": self.subscription.subscription_id,
            "result": {
                "data": [{"some_col": 101}],
                "meta": [{"name": "count", "type": "UInt64"}],
            },
            "request": {
                "some": "data",
                "query": """MATCH (metrics_counters) SELECT sum(value) AS value BY
                        tags[3] WHERE org_id = 1 AND project_id IN tuple(1) AND metric_id = 16
                        AND tags[3] IN tuple(13, 4)""",
            },
            "entity": "metrics_counters",
            "timestamp": "2020-01-01T01:23:45.1234",
        },
    }
    self.producer.produce(self.topic, json.dumps(message))
    self.producer.flush()
    def active_incident():
        return Incident.objects.filter(
            type=IncidentType.ALERT_TRIGGERED.value, alert_rule=self.rule
        ).exclude(status=IncidentStatus.CLOSED.value)
    original_callback = subscriber_registry[INCIDENTS_SNUBA_SUBSCRIPTION_TYPE]
    def shutdown_callback(*args, **kwargs):
        # We want to just exit after the callback so that we can see the result of
        # processing.
        original_callback(*args, **kwargs)
        consumer.signal_shutdown()
    subscriber_registry[INCIDENTS_SNUBA_SUBSCRIPTION_TYPE] = shutdown_callback
    with self.feature(["organizations:incidents", "organizations:performance-view"]):
        assert not active_incident().exists()
        with self.tasks(), self.capture_on_commit_callbacks(execute=True):
            consumer.run()
        assert active_incident().exists()
    assert len(mail.outbox) == 1
    handler = EmailActionHandler(self.action, active_incident().get(), self.project)
    incident_activity = (
        IncidentActivity.objects.filter(incident=handler.incident).order_by("-id").first()
    )
    message_builder = handler.build_message(
        generate_incident_trigger_email_context(
            handler.project,
            handler.incident,
            handler.action.alert_rule_trigger,
            TriggerStatus.ACTIVE,
            IncidentStatus.CRITICAL,
            notification_uuid=str(incident_activity.notification_uuid),
        ),
        TriggerStatus.ACTIVE,
        self.user.id,
    )
    out = mail.outbox[0]
    assert isinstance(out, mail.EmailMultiAlternatives)
    assert out.to == [self.user.email]
    assert out.subject == message_builder.subject
    built_message = message_builder.build(self.user.email)
    assert out.body == built_message.body