Make partial error generic and move it to the consumer package

Signed-off-by: Pavol Loffay <[email protected]>

Author: pavolloffay

processor/queuedprocessor/partial_error.go renamed to consumer/consumererror/partialerror.go

@@ -12,20 +12,27 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package queuedprocessor
+package consumererror
 
 import "go.opentelemetry.io/collector/consumer/pdata"
 
-type partialRetryError struct {
+// PartialError can be used to signalize that a subset of received data failed to be processed or send.
+// The preceding components in the pipeline can use this information for partial retries.
+type PartialError struct {
 	error
-	toRetry pdata.Traces
+	failed pdata.Traces
 }
 
-// NewPartialRetryError creates an error that encapsulates data that should be retried.
-// Use this error type only when a subset of received data set should bre retried.
-func NewPartialRetryError(err error, toRetry pdata.Traces) error {
-	return partialRetryError{
-		error:   err,
-		toRetry: toRetry,
+// PartialTracesError creates PartialError for failed traces.
+// Use this error type only when a subset of received data set failed to be processed or sent.
+func PartialTracesError(err error, failed pdata.Traces) error {
+	return PartialError{
+		error:  err,
+		failed: failed,
 	}
 }
+
+// GetTraces returns failed traces.
+func (err PartialError) GetTraces() pdata.Traces {
+	return err.failed
+}

processor/queuedprocessor/partial_error_test.go renamed to consumer/consumererror/partialerror_test.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package queuedprocessor
+package consumererror
 
 import (
 	"fmt"
@@ -26,7 +26,7 @@ import (
 func TestPartialError(t *testing.T) {
 	td := testdata.GenerateTraceDataOneSpan()
 	err := fmt.Errorf("some error")
-	partialErr := NewPartialRetryError(err, td)
+	partialErr := PartialTracesError(err, td)
 	assert.Equal(t, err.Error(), partialErr.Error())
-	assert.Equal(t, td, partialErr.(partialRetryError).toRetry)
+	assert.Equal(t, td, partialErr.(PartialError).failed)
 }

consumer/consumererror/consumererror.go renamed to consumer/consumererror/permanenterror.go

@@ -193,8 +193,8 @@ func (sp *queuedSpanProcessor) processItemFromQueue(item *queueItem) {
 		sp.logger.Error("Failed to process batch, discarding", zap.Int("batch-size", allSpansCount))
 		sp.onItemDropped(item, spanCountStats)
 	} else {
-		if partialErr, isPartial := err.(partialRetryError); isPartial {
-			item.td = partialErr.toRetry
+		if partialErr, isPartial := err.(consumererror.PartialError); isPartial {
+			item.td = partialErr.GetTraces()
 			spanCountStats = processor.NewSpanCountStats(item.td, sp.name)
 			allSpansCount = spanCountStats.GetAllSpansCount()
 		}

consumer/consumererror/consumererror_test.go renamed to consumer/consumererror/permanenterror_test.go

@@ -73,7 +73,7 @@ func TestQueuedProcessor_noEnqueueOnPermanentError(t *testing.T) {
 }
 
 func TestQueueProcessorPartialError(t *testing.T) {
-	partialErr := partialRetryError{toRetry: testdata.GenerateTraceDataTwoSpansSameResource()}
+	partialErr := consumererror.PartialTracesError(fmt.Errorf("some error"), testdata.GenerateTraceDataTwoSpansSameResource())
 	td := testdata.GenerateTraceDataTwoSpansSameResource()
 	c := &waitGroupTraceConsumer{
 		consumeTraceDataError: partialErr,
@@ -93,7 +93,7 @@ func TestQueueProcessorPartialError(t *testing.T) {
 	exporterData := c.getData()
 	assert.Equal(t, 2, len(exporterData))
 	assert.Equal(t, td, exporterData[0])
-	assert.Equal(t, partialErr.toRetry, exporterData[1])
+	assert.Equal(t, partialErr.(consumererror.PartialError).GetTraces(), exporterData[1])
 }
 
 type waitGroupTraceConsumer struct {