Sync from internal repo through 6035345

idk/Makefile

@@ -227,11 +227,11 @@ test-run: testenv vendor
 
 test-run-race: testenv vendor
 	$(DOCKER_COMPOSE) build idk-test
-	$(DOCKER_COMPOSE) run -T idk-test bash -c "set -o pipefail; go test -v -mod=vendor -race -covermode=atomic -tags=dynamic $(TPKG) -coverpkg=$(TPKG) -timeout=30m -json -coverprofile=/testdata/$(PROJECT)_coverage.out  | tee /testdata/$(PROJECT)_report.out"
+	$(DOCKER_COMPOSE) run -T idk-test bash -c "set -o pipefail; go test -v -mod=vendor -race -covermode=atomic -tags=dynamic $(TPKG) -coverpkg=$(TPKG) -timeout=30m -json -coverprofile=/testdata/$(PROJECT)_race_coverage.out  | tee /testdata/$(PROJECT)_report.out"
 
 test-run-kafka-sasl: testenv vendor
 	$(DOCKER_COMPOSE) build idk-test
-	$(DOCKER_COMPOSE) run -T idk-test bash -c "set -o pipefail; go test -v --tags=kafka_sasl -mod=vendor -race -timeout=30m $(TPKG) -covermode=atomic -coverpkg=$(TPKG) -json -coverprofile=/testdata/$(PROJECT)_coverage.out  | tee /testdata/$(PROJECT)_report.out"
+	$(DOCKER_COMPOSE) run -T idk-test bash -c "set -o pipefail; go test -v --tags=kafka_sasl -mod=vendor -race -timeout=30m $(TPKG) -covermode=atomic -coverpkg=$(TPKG) -json -coverprofile=/testdata/$(PROJECT)_sasl_coverage.out  | tee /testdata/$(PROJECT)_report.out"
 
 .pulled:
 	$(DOCKER_COMPOSE) pull 

idk/cmd/molecula-consumer-kinesis/main.go

@@ -9,18 +9,24 @@ import (
 	"github.com/featurebasedb/featurebase/v3/logger"
 )
 
-func logFailure(errorType kinesis.ErrorType, m *kinesis.Main, v interface{}) {
+func logError(m *kinesis.Main, err error) {
 	log := m.Log()
 
 	if log == nil {
 		log = logger.NewStandardLogger(os.Stderr)
 	}
+	log.Errorf("Error running command: %s", err)
 
-	if errorType == kinesis.RecoverableErrorType {
-		log.Errorf("Error running command: %+v", v)
-	} else {
-		log.Panicf("Panic running command: %+v", v)
+}
+
+func logPanic(m *kinesis.Main, v interface{}) {
+	log := m.Log()
+
+	if log == nil {
+		log = logger.NewStandardLogger(os.Stderr)
 	}
+	log.Panicf("Panic running command: %+v", v)
+
 }
 
 func main() {
@@ -33,7 +39,7 @@ func main() {
 	// Capture any panic and log it before dying.
 	defer func() {
 		if r := recover(); r != nil {
-			logFailure(kinesis.PanicErrorType, m, r)
+			logPanic(m, r)
 			os.Exit(1)
 		}
 	}()
@@ -44,7 +50,7 @@ func main() {
 	}
 
 	if err := m.Run(); err != nil {
-		logFailure(kinesis.RecoverableErrorType, m, err)
+		logError(m, err)
 		os.Exit(1)
 	}
 }

idk/ingest.go

@@ -92,6 +92,7 @@ type Main struct {
 	AllowIntOutOfRange       bool          `help:"Allow ingest to continue when it encounters out of range integers in IntFields. (default false)"`
 	AllowDecimalOutOfRange   bool          `help:"Allow ingest to continue when it encounters out of range decimals in DecimalFields. (default false)"`
 	AllowTimestampOutOfRange bool          `help:"Allow ingest to continue when it encounters out of range timestamps in TimestampFields. (default false)"`
+	SkipBadRows              int           `help:"If you fail to process the first n rows without processing one successfully, fail."`
 
 	UseShardTransactionalEndpoint bool `flag:"use-shard-transactional-endpoint" help:"Use alternate import endpoint. Currently unstable/testing"`
 
@@ -325,6 +326,8 @@ func (m *Main) ingest(ctx context.Context, source Source, nexter IDAllocator, so
 	var recordizers []Recordizer
 	var prevRec Record
 	var row *pilosaclient.Row
+	var errorCounter int // keeps track of consecuitive errors across records
+	var anyRecordSuccessful bool
 	if m.progress != nil {
 		source = m.progress.Track(source)
 	}
@@ -480,6 +483,7 @@ initialFetch:
 			}
 		}
 
+		rowHasError := false
 		for _, rdz := range recordizers {
 			err = rdz(data, row)
 			if err != nil {
@@ -487,11 +491,32 @@ initialFetch:
 				// excludes 'out of range' errors so that ingest can continue importing the rest
 				// of the records
 				if !m.allowError(err) {
-					return err
+					rowHasError = true
+					// must return error and exit idk when SkipBadRows is not defined (or set to 0)
+					if m.SkipBadRows == 0 {
+						return err
+					} else {
+						// will handle this error based on errorCounter in the later if block
+						m.Log().Errorf("Bad record: +%v, reason: %v\n", row, err)
+						break
+					}
 				}
 				m.Log().Errorf(err.Error())
 			}
 		}
+
+		if !anyRecordSuccessful && rowHasError {
+			// We cannot allow a certain number of consecutive errors in the beginning of ingest.
+			errorCounter++
+			if errorCounter > m.SkipBadRows {
+				return errors.Wrapf(err, "consecutive bad records exceeded limit, errorCounter: %d\n", errorCounter) // wraps the current recordizer error
+			}
+			err = nil // already logged the recordizer error, no need to propagate the err
+		} else {
+			anyRecordSuccessful = true // after this is set true, we will skip any bad rows in the future
+			err = nil
+		}
+
 		if nexter != nil { // add ID if no id field specified
 			if batchStart {
 				rerr := nexter.Reserve(ctx, uint64(m.BatchSize))
@@ -522,8 +547,13 @@ initialFetch:
 			}
 			row.ID = id
 		}
-		err = batch.Add(*row)
-		m.stats.Count(MetricIngesterRowsAdded, 1, 1)
+
+		// skip bad rows only
+		if !rowHasError {
+			err = batch.Add(*row)
+			m.stats.Count(MetricIngesterRowsAdded, 1, 1)
+		}
+
 		if err == pilosaclient.ErrBatchNowFull || err == pilosaclient.ErrBatchNowStale {
 			batchLen := batch.Len()
 			err = m.importBatch(batch)

idk/ingest_test.go

@@ -194,6 +194,64 @@ func TestIngestSignedIntBoolField(t *testing.T) {
 	}
 }
 
+//The following two tests are used to test the functionality of a new feature in which we skip bad records coming into idk and log them.
+//First function checks that we can successfully skip some bad rows coming into idk
+//Second function checks whether we get error if we have more bad records than acceptable errors by idk mentioned by SkipBadRows parameter.
+
+func skipBadRowsTestSource() *testSource {
+	ts := newTestSource([]Field{StringField{NameVal: "rcid"}, SignedIntBoolKeyField{NameVal: "svals"}},
+		[][]interface{}{
+			{"c", "badrecord1"},
+			{"d", "badrecord2"},
+			{"a", "badrecord3"},
+			{"x", int64(66)},
+			{"b", int64(11)},
+			{"b", int64(22)},
+			{"b", int64(-32)},
+			{"b", int64(-44)},
+			{"b", "badrecord4"},
+			{"b", int64(11)},
+			{"b", int64(7)},
+			{"c", int64(5)},
+		})
+	return ts
+
+}
+func ingesterCreationForSkipBadRowsTest(skipbadrows int) *Main {
+	ts := skipBadRowsTestSource()
+	ingester := NewMain()
+	configureTestFlags(ingester)
+	ingester.NewSource = func() (Source, error) { return ts, nil }
+	rand.Seed(time.Now().UTC().UnixNano())
+	ingester.Index = fmt.Sprintf("ingestint%d", rand.Intn(100000))
+	ingester.BatchSize = 2
+	ingester.SkipBadRows = skipbadrows
+	ingester.PrimaryKeyFields = []string{"rcid"}
+	return ingester
+}
+func TestSkipBadRowsFunctionality(t *testing.T) {
+
+	ingester := ingesterCreationForSkipBadRowsTest(3)
+
+	err := ingester.Run()
+	if err != nil {
+		t.Fatalf("%s: %v", idktest.ErrRunningIngest, err)
+	}
+}
+
+func TestSkipBadRowsFunctionalityWhenErrorCountIsMore(t *testing.T) {
+
+	ingester := ingesterCreationForSkipBadRowsTest(1)
+
+	err := ingester.Run()
+	if err == nil {
+		t.Fatalf("%s: %v", idktest.ErrRunningIngest, err)
+	}
+	if !strings.Contains(err.Error(), "consecutive bad records exceeded limit") {
+		t.Fatalf("did not receive expected error from idk %v", err.Error())
+	}
+}
+
 // TestSingleBoolClear essentially creates an import batch which
 // clears a bit in a particular fragment without setting a bit in that
 // same fragment.  There's a potential optimization in the pilosa client