Treat syntax error as non internal

Author: eminano

pkg/stream/stream_run.go

@@ -211,6 +211,7 @@ func Run(ctx context.Context, logger loglib.Logger, config *Config, instrumentat
 			return err
 		}
 		defer pgBatchWriter.Close()
+		logger.Info("starting postgres batch writer...")
 
 		processor = pgBatchWriter
 
@@ -222,6 +223,7 @@ func Run(ctx context.Context, logger loglib.Logger, config *Config, instrumentat
 		logger.Info("adding transformation layer to processor...")
 		transformer, err := transformer.New(config.Processor.Transformer, processor, transformer.WithLogger(logger))
 		if err != nil {
+			logger.Error(err, "creating transformer layer")
 			return err
 		}
 		processor = transformer

pkg/wal/processor/postgres/postgres_batch_writer.go

@@ -242,9 +242,11 @@ func (w *BatchWriter) execQueries(ctx context.Context, queries []*query) ([]*que
 func (w *BatchWriter) isInternalError(err error) bool {
 	var errRelationDoesNotExist *pglib.ErrRelationDoesNotExist
 	var errConstraintViolation *pglib.ErrConstraintViolation
+	var errSyntaxError *pglib.ErrSyntaxError
 	switch {
 	case errors.As(err, &errRelationDoesNotExist),
-		errors.As(err, &errConstraintViolation):
+		errors.As(err, &errConstraintViolation),
+		errors.As(err, &errSyntaxError):
 		return false
 	default:
 		return true