diff --git a/metricbeat/module/windows/perfmon/pdh_integration_windows_test.go b/metricbeat/module/windows/perfmon/pdh_integration_windows_test.go
index 18ecaccb493..04e4a5b7b72 100644
--- a/metricbeat/module/windows/perfmon/pdh_integration_windows_test.go
+++ b/metricbeat/module/windows/perfmon/pdh_integration_windows_test.go
@@ -58,11 +58,11 @@ func TestData(t *testing.T) {
 		},
 	}
 
-	ms := mbtest.NewReportingMetricSetV2(t, config)
-	mbtest.ReportingFetchV2(ms)
+	ms := mbtest.NewReportingMetricSetV2Error(t, config)
+	mbtest.ReportingFetchV2Error(ms)
 	time.Sleep(60 * time.Millisecond)
 
-	events, errs := mbtest.ReportingFetchV2(ms)
+	events, errs := mbtest.ReportingFetchV2Error(ms)
 	if len(errs) > 0 {
 		t.Fatal(errs)
 	}
@@ -87,11 +87,11 @@ func TestCounterWithNoInstanceName(t *testing.T) {
 		},
 	}
 
-	ms := mbtest.NewReportingMetricSetV2(t, config)
-	mbtest.ReportingFetchV2(ms)
+	ms := mbtest.NewReportingMetricSetV2Error(t, config)
+	mbtest.ReportingFetchV2Error(ms)
 	time.Sleep(60 * time.Millisecond)
 
-	events, errs := mbtest.ReportingFetchV2(ms)
+	events, errs := mbtest.ReportingFetchV2Error(ms)
 	if len(errs) > 0 {
 		t.Fatal(errs)
 	}
diff --git a/metricbeat/module/windows/perfmon/perfmon.go b/metricbeat/module/windows/perfmon/perfmon.go
index a73dbb57c46..93645ee77b4 100644
--- a/metricbeat/module/windows/perfmon/perfmon.go
+++ b/metricbeat/module/windows/perfmon/perfmon.go
@@ -88,17 +88,21 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	}, nil
 }
 
-func (m *MetricSet) Fetch(report mb.ReporterV2) {
+// Fetch fetches events and reports them upstream
+func (m *MetricSet) Fetch(report mb.ReporterV2) error {
 	events, err := m.reader.Read()
 	if err != nil {
-		m.log.Debugw("Failed reading counters", "error", err)
-		err = errors.Wrap(err, "failed reading counters")
-		report.Error(err)
+		return errors.Wrap(err, "failed reading counters")
 	}
 
 	for _, event := range events {
-		report.Event(event)
+		isOpen := report.Event(event)
+		if !isOpen {
+			break
+		}
 	}
+
+	return nil
 }
 
 // Close will be called when metricbeat is stopped, should close the query.