Repository: cloudspannerecosystem/harbourbridge
Pull request #331 (2022-07-06 03:20:51)

Adding Dockerfile of HarbourBridge's Docker image
This PR consists of -  
-  The `Dockerfile` of HarbourBridge's Docker image. 

This file is necessary to enable Google Cloud Build to build HarbourBridge's Docker images and put them in the Google Container Registry. 

The built images 
- Will support all the migration modes of HarbourBridge - 
  - schema
  - data
  - schema-and-data

- Will support both forms of source databases that HarbourBridge does - 
  - Dump files in the host's file system.
  - Remotely hosted databases.

The `Dockerfile` consists of -
- Build Stage : This stage uses the HarbourBridge source code,  its go.mod dependencies and Golang to compile the software into a binary file `harbourbridge`. This binary is copied into the release stage.
- Release Stage : The image built in this stage will be available to the users. It packages the HarbourBridge binary file `harbourbridge` and GCloud SDK together. 
  - The binary will get executed to perform migrations from the source databases to Spanner.
  - GCloud SDK will authenticate the host's GCloud account and put the migrated database in the relevant Spanner instance.
  - GCloud SDK needs python and an OS to function, thus `Python3` and `Ubuntu OS` are also part of the image.
  


Patch:
diff --git a/Dockerfile b/Dockerfile
new file mode 100644
index 0000000..6f9e661
--- /dev/null
+++ b/Dockerfile
@@ -0,0 +1,63 @@
+# Copyright 2022 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+# syntax=docker/dockerfile:1
+
+############################# BUILD ###############################
+# This stage builds the HarbourBridge binary using its source code.
+
+# Add golang as the base image for the build stage.
+FROM golang:1.18.3 AS build
+
+# Set the default work directory as “/hbBin” in the container.
+WORKDIR /harbourbridge_bin
+
+# Copy HarbourBridge’s source code into the working directory.
+COPY . ./
+# Download HarbourBridge’s dependencies mentioned in the go.mod file.
+RUN go mod tidy 
+# Compile the source code and store the binary in the work directory /hbBin.
+RUN make build-static  
+
+########################### RELEASE ###############################
+# This stage consists of the GCloud SDK and the HarbourBridge binary.
+
+# Add the Ubuntu base image from the Google Container Registry.
+FROM gcr.io/cloud-marketplace/google/ubuntu1804@sha256:d9fe979ab72ff02f5b5d95c59cda6681aa73039fbf9b46596935267cf26c692e
+# Set the default working directory.
+# This directory binds with the host directory that is to store the output files generated by HarbourBridge.
+WORKDIR /harbourbridge/harbour_bridge_output
+# Create directory to store the HarbourBridge binary. 
+RUN mkdir /harbourbridge/bin
+# Create directory to store the dump files. This directory binds with the host directory that contains the dump files of the source database.
+RUN mkdir /harbourbridge/source_dump
+# Copy the HarbourBridge binary from the build stage's hbBin directory to the release stage's harbourbridge/bin directory.
+COPY --from=build /harbourbridge_bin/harbourbridge /harbourbridge/bin
+
+# Add Python3 to support GCloud SDK.
+RUN apt-get update
+RUN apt-get install -y python3.10
+
+# Download the gcloud package.
+RUN curl https://dl.google.com/dl/cloudsdk/channels/rapid/downloads/google-cloud-cli-391.0.0-linux-x86_64.tar.gz > /tmp/google-cloud-sdk.tar.gz
+
+# Install the GCloud package.
+RUN mkdir -p /usr/local/gcloud \
+  && tar -C /usr/local/gcloud -xf /tmp/google-cloud-sdk.tar.gz \
+  && /usr/local/gcloud/google-cloud-sdk/install.sh
+
+# Add gcloud to $PATH.
+ENV PATH $PATH:/usr/local/gcloud/google-cloud-sdk/bin
+
diff --git a/cmd/data.go b/cmd/data.go
index b602035..4f9fbc2 100644
--- a/cmd/data.go
+++ b/cmd/data.go
@@ -26,10 +26,12 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
 	"github.com/google/subcommands"
 	"github.com/google/uuid"
+	"go.uber.org/zap"
 )
 
 // DataCmd struct with flags.
@@ -42,6 +44,8 @@ type DataCmd struct {
 	sessionJSON     string
 	filePrefix      string // TODO: move filePrefix to global flags
 	writeLimit      int64
+	dryRun          bool
+	logLevel        string
 }
 
 // Name returns the name of operation.
@@ -75,6 +79,8 @@ func (cmd *DataCmd) SetFlags(f *flag.FlagSet) {
 	f.BoolVar(&cmd.skipForeignKeys, "skip-foreign-keys", false, "Skip creating foreign keys after data migration is complete (ddl statements for foreign keys can still be found in the downloaded schema.ddl.txt file and the same can be applied separately)")
 	f.StringVar(&cmd.filePrefix, "prefix", "", "File prefix for generated files")
 	f.Int64Var(&cmd.writeLimit, "write-limit", defaultWritersLimit, "Write limit for writes to spanner")
+	f.BoolVar(&cmd.dryRun, "dry-run", false, "To validate the syntax of the command by running it in an air-gapped manner, such that no network calls are made.")
+	f.StringVar(&cmd.logLevel, "log-level", "INFO", "Configure the logging level for the command (INFO, DEBUG), defaults to INFO")
 }
 
 func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface{}) subcommands.ExitStatus {
@@ -83,9 +89,20 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 	var err error
 	defer func() {
 		if err != nil {
-			fmt.Printf("FATAL error: %v\n", err)
+			logger.Log.Fatal("FATAL error", zap.Error(err))
 		}
 	}()
+	if cmd.dryRun {
+		fmt.Print("--dry-run flag is not implemented")
+		return subcommands.ExitFailure
+	}
+	err = logger.InitializeLogger(cmd.logLevel)
+	if err != nil {
+		fmt.Println("Error initialising logger, did you specify a valid log-level? [DEBUG, INFO, WARN, ERROR, FATAL]", err)
+		return subcommands.ExitFailure
+	}
+	defer logger.Log.Sync()
+
 	conv := internal.MakeConv()
 
 	sourceProfile, err := profiles.NewSourceProfile(cmd.sourceProfile, cmd.source)
diff --git a/cmd/schema.go b/cmd/schema.go
index eb02082..3afd93c 100644
--- a/cmd/schema.go
+++ b/cmd/schema.go
@@ -26,8 +26,10 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/google/subcommands"
+	"go.uber.org/zap"
 )
 
 // SchemaCmd struct with flags.
@@ -37,6 +39,7 @@ type SchemaCmd struct {
 	target        string
 	targetProfile string
 	filePrefix    string // TODO: move filePrefix to global flags
+	logLevel      string
 }
 
 // Name returns the name of operation.
@@ -67,6 +70,7 @@ func (cmd *SchemaCmd) SetFlags(f *flag.FlagSet) {
 	f.StringVar(&cmd.target, "target", "Spanner", "Specifies the target DB, defaults to Spanner (accepted values: `Spanner`)")
 	f.StringVar(&cmd.targetProfile, "target-profile", "", "Flag for specifying connection profile for target database e.g., \"dialect=postgresql\"")
 	f.StringVar(&cmd.filePrefix, "prefix", "", "File prefix for generated files")
+	f.StringVar(&cmd.logLevel, "log-level", "INFO", "Configure the logging level for the command (INFO, DEBUG), defaults to INFO")
 }
 
 func (cmd *SchemaCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface{}) subcommands.ExitStatus {
@@ -75,9 +79,15 @@ func (cmd *SchemaCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interfa
 	var err error
 	defer func() {
 		if err != nil {
-			fmt.Printf("FATAL error: %v\n", err)
+			logger.Log.Fatal("FATAL error", zap.Error(err))
 		}
 	}()
+	err = logger.InitializeLogger(cmd.logLevel)
+	if err != nil {
+		fmt.Println("Error initialising logger, did you specify a valid log-level? [DEBUG, INFO, WARN, ERROR, FATAL]", err)
+		return subcommands.ExitFailure
+	}
+	defer logger.Log.Sync()
 
 	sourceProfile, err := profiles.NewSourceProfile(cmd.sourceProfile, cmd.source)
 	if err != nil {
diff --git a/cmd/schema_and_data.go b/cmd/schema_and_data.go
index afacfb9..282e2df 100644
--- a/cmd/schema_and_data.go
+++ b/cmd/schema_and_data.go
@@ -26,10 +26,12 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
 	"github.com/google/subcommands"
 	"github.com/google/uuid"
+	"go.uber.org/zap"
 )
 
 // SchemaAndDataCmd struct with flags.
@@ -41,6 +43,8 @@ type SchemaAndDataCmd struct {
 	skipForeignKeys bool
 	filePrefix      string // TODO: move filePrefix to global flags
 	writeLimit      int64
+	dryRun          bool
+	logLevel        string
 }
 
 // Name returns the name of operation.
@@ -73,6 +77,8 @@ func (cmd *SchemaAndDataCmd) SetFlags(f *flag.FlagSet) {
 	f.BoolVar(&cmd.skipForeignKeys, "skip-foreign-keys", false, "Skip creating foreign keys after data migration is complete (ddl statements for foreign keys can still be found in the downloaded schema.ddl.txt file and the same can be applied separately)")
 	f.StringVar(&cmd.filePrefix, "prefix", "", "File prefix for generated files")
 	f.Int64Var(&cmd.writeLimit, "write-limit", defaultWritersLimit, "Write limit for writes to spanner")
+	f.BoolVar(&cmd.dryRun, "dry-run", false, "To validate the syntax of the command by running it in an air-gapped manner, such that no network calls are made.")
+	f.StringVar(&cmd.logLevel, "log-level", "INFO", "Configure the logging level for the command (INFO, DEBUG), defaults to INFO")
 }
 
 func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface{}) subcommands.ExitStatus {
@@ -81,9 +87,19 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 	var err error
 	defer func() {
 		if err != nil {
-			fmt.Printf("FATAL error: %v\n", err)
+			logger.Log.Fatal("FATAL error", zap.Error(err))
 		}
 	}()
+	if cmd.dryRun {
+		fmt.Print("--dry-run flag is not implemented")
+		return subcommands.ExitFailure
+	}
+	err = logger.InitializeLogger(cmd.logLevel)
+	if err != nil {
+		fmt.Println("Error initialising logger, did you specify a valid log-level? [DEBUG, INFO, WARN, ERROR, FATAL]", err)
+		return subcommands.ExitFailure
+	}
+	defer logger.Log.Sync()
 
 	sourceProfile, err := profiles.NewSourceProfile(cmd.sourceProfile, cmd.source)
 	if err != nil {
diff --git a/common/utils/utils.go b/common/utils/utils.go
index 4081b2d..139ae8c 100644
--- a/common/utils/utils.go
+++ b/common/utils/utils.go
@@ -237,6 +237,12 @@ func getInstances(ctx context.Context, project string) ([]string, error) {
 }
 
 func GetPassword() string {
+	calledFromGCloud := os.Getenv("GCLOUD_HB_PLUGIN")
+	if strings.EqualFold(calledFromGCloud, "true") {
+		fmt.Println("\n Please specify password in enviroment variables (recommended) or --source-profile " +
+			"(not recommended) while using HarbourBridge from gCloud CLI.")
+		return ""
+	}
 	fmt.Print("Enter Password: ")
 	bytePassword, err := terminal.ReadPassword(int(syscall.Stdin))
 	if err != nil {
diff --git a/conversion/conversion.go b/conversion/conversion.go
index 1dbd0ae..3843eea 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -44,14 +44,11 @@ import (
 	"github.com/aws/aws-sdk-go/aws/session"
 	dydb "github.com/aws/aws-sdk-go/service/dynamodb"
 	"github.com/aws/aws-sdk-go/service/dynamodbstreams"
-	adminpb "google.golang.org/genproto/googleapis/spanner/admin/database/v1"
-	"google.golang.org/grpc/metadata"
-	"google.golang.org/protobuf/proto"
-
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/common/metrics"
 	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/csv"
@@ -62,6 +59,10 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/sources/sqlserver"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/writer"
+	"go.uber.org/zap"
+	adminpb "google.golang.org/genproto/googleapis/spanner/admin/database/v1"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/protobuf/proto"
 )
 
 var (
@@ -390,6 +391,7 @@ func getSeekable(f *os.File) (*os.File, int64, error) {
 		return f, n, err
 	}
 	internal.VerbosePrintln("Creating a tmp file with a copy of stdin because stdin is not seekable.")
+	logger.Log.Debug("Creating a tmp file with a copy of stdin because stdin is not seekable.")
 
 	// Create file in os.TempDir. Its not clear this is a good idea e.g. if the
 	// pg_dump/mysqldump output is large (tens of GBs) and os.TempDir points to a directory
@@ -426,15 +428,28 @@ func VerifyDb(ctx context.Context, adminClient *database.DatabaseAdminClient, db
 }
 
 // CheckExistingDb checks whether the database with dbURI exists or not.
+// If API call doesn't respond then user is informed after every 5 minutes on command line.
 func CheckExistingDb(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string) (bool, error) {
-	_, err := adminClient.GetDatabase(ctx, &adminpb.GetDatabaseRequest{Name: dbURI})
-	if err != nil {
-		if utils.ContainsAny(strings.ToLower(err.Error()), []string{"database not found"}) {
-			return false, nil
+	gotResponse := make(chan bool)
+	var err error
+	go func() {
+		_, err = adminClient.GetDatabase(ctx, &adminpb.GetDatabaseRequest{Name: dbURI})
+		gotResponse <- true
+	}()
+	for {
+		select {
+		case <-time.After(5 * time.Minute):
+			fmt.Println("WARNING! API call not responding: make sure that spanner api endpoint is configured properly")
+		case <-gotResponse:
+			if err != nil {
+				if utils.ContainsAny(strings.ToLower(err.Error()), []string{"database not found"}) {
+					return false, nil
+				}
+				return false, fmt.Errorf("can't get database info: %s", err)
+			}
+			return true, nil
 		}
-		return false, fmt.Errorf("can't get database info: %s", err)
 	}
-	return true, nil
 }
 
 // ValidateDDL verifies if an existing DB's ddl follows what is supported by harbourbridge. Currently,
@@ -586,6 +601,8 @@ Recommended value is between 20-30.`)
 				workers <- workerID
 			}()
 			internal.VerbosePrintf("Submitting new FK create request: %s\n", fkStmt)
+			logger.Log.Debug("Submitting new FK create request", zap.String("fkStmt", fkStmt))
+
 			op, err := adminClient.UpdateDatabaseDdl(ctx, &adminpb.UpdateDatabaseDdlRequest{
 				Database:   dbURI,
 				Statements: []string{fkStmt},
@@ -601,6 +618,7 @@ Recommended value is between 20-30.`)
 				return
 			}
 			internal.VerbosePrintln("Updated schema with statement: " + fkStmt)
+			logger.Log.Debug("Updated schema with statement", zap.String("fkStmt", fkStmt))
 		}(fkStmt, workerID)
 	}
 	// Wait for all the goroutines to finish.
diff --git a/go.mod b/go.mod
index df818ca..8f5eb59 100644
--- a/go.mod
+++ b/go.mod
@@ -32,6 +32,7 @@ require (
 	github.com/pingcap/tidb/parser v0.0.0-20220411093434-32b9c14779c2
 	github.com/sijms/go-ora/v2 v2.2.17
 	github.com/stretchr/testify v1.7.0
+	go.uber.org/zap v1.21.0
 	golang.org/x/crypto v0.0.0-20220214200702-86341886e292
 	golang.org/x/net v0.0.0-20220325170049-de3da57026de
 	golang.org/x/tools v0.1.11-0.20220316014157-77aa08bb151a // indirect
diff --git a/internal/convert.go b/internal/convert.go
index 8f66667..475fb3a 100644
--- a/internal/convert.go
+++ b/internal/convert.go
@@ -18,9 +18,11 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+	"go.uber.org/zap"
 )
 
 // Conv contains all schema and data conversion state.
@@ -209,6 +211,8 @@ func (conv *Conv) WriteRow(srcTable, spTable string, spCols []string, spVals []i
 	if conv.dataSink == nil {
 		msg := "Internal error: ProcessDataRow called but dataSink not configured"
 		VerbosePrintf("%s\n", msg)
+		logger.Log.Debug("Internal error: ProcessDataRow called but dataSink not configured")
+
 		conv.Unexpected(msg)
 		conv.StatsAddBadRow(srcTable, conv.DataMode())
 	} else {
@@ -347,6 +351,8 @@ func (conv *Conv) buildPrimaryKey(spTable string) string {
 // because we process dump data twice.
 func (conv *Conv) Unexpected(u string) {
 	VerbosePrintf("Unexpected condition: %s\n", u)
+	logger.Log.Debug("Unexpected condition", zap.String("condition", u))
+
 	// Limit size of unexpected map. If over limit, then only
 	// update existing entries.
 	if _, ok := conv.Stats.Unexpected[u]; ok || len(conv.Stats.Unexpected) < 1000 {
@@ -395,6 +401,7 @@ func (conv *Conv) getStatementStat(s string) *statementStat {
 func (conv *Conv) SkipStatement(stmtType string) {
 	if conv.SchemaMode() { // Record statement stats on first pass only.
 		VerbosePrintf("Skipping statement: %s\n", stmtType)
+		logger.Log.Debug("Skipping statement", zap.String("stmtType", stmtType))
 		conv.getStatementStat(stmtType).Skip++
 	}
 }
@@ -403,6 +410,7 @@ func (conv *Conv) SkipStatement(stmtType string) {
 func (conv *Conv) ErrorInStatement(stmtType string) {
 	if conv.SchemaMode() { // Record statement stats on first pass only.
 		VerbosePrintf("Error processing statement: %s\n", stmtType)
+		logger.Log.Debug("Error processing statement", zap.String("stmtType", stmtType))
 		conv.getStatementStat(stmtType).Error++
 	}
 }
diff --git a/internal/mapping.go b/internal/mapping.go
index 386176a..48ff178 100644
--- a/internal/mapping.go
+++ b/internal/mapping.go
@@ -19,6 +19,7 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
@@ -46,6 +47,7 @@ func GetSpannerTable(conv *Conv, srcTable string) (string, error) {
 	spTable := getSpannerID(conv, srcTable)
 	if spTable != srcTable {
 		VerbosePrintf("Mapping source DB table %s to Spanner table %s\n", srcTable, spTable)
+		logger.Log.Debug(fmt.Sprintf("Mapping source DB table %s to Spanner table %s\n", srcTable, spTable))
 	}
 	conv.ToSpanner[srcTable] = NameAndCols{Name: spTable, Cols: make(map[string]string)}
 	conv.ToSource[spTable] = NameAndCols{Name: srcTable, Cols: make(map[string]string)}
@@ -118,6 +120,7 @@ func GetSpannerCol(conv *Conv, srcTable, srcCol string, mustExist bool) (string,
 	}
 	if spCol != srcCol {
 		VerbosePrintf("Mapping source DB col %s (table %s) to Spanner col %s\n", srcCol, srcTable, spCol)
+		logger.Log.Debug(fmt.Sprintf("Mapping source DB col %s (table %s) to Spanner col %s\n", srcCol, srcTable, spCol))
 	}
 	conv.ToSpanner[srcTable].Cols[srcCol] = spCol
 	conv.ToSource[sp.Name].Cols[spCol] = srcCol
diff --git a/internal/progress.go b/internal/progress.go
index 627150f..f26829f 100644
--- a/internal/progress.go
+++ b/internal/progress.go
@@ -20,6 +20,9 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
+	"go.uber.org/zap"
 )
 
 // Progress provides console progress functionality. i.e. it reports what
@@ -87,6 +90,7 @@ func (p *Progress) reportPct(firstCall bool) {
 		fmt.Printf("%s: %2d%%\n", p.message, p.pct)
 		return
 	}
+	logger.Log.Debug(p.message, zap.Int("Progress", p.pct))
 	if firstCall {
 		fmt.Printf("%s: %2d%%", p.message, p.pct)
 	} else {
@@ -102,6 +106,7 @@ func (p *Progress) reportFraction(firstCall bool) {
 		fmt.Printf("%s: %d/%d\n", p.message, p.progress, p.total)
 		return
 	}
+	logger.Log.Debug(p.message, zap.Float32("Progress", float32(p.progress/p.total)))
 	if firstCall {
 		fmt.Printf("%s: %d/%d", p.message, p.progress, p.total)
 	} else {
diff --git a/logger/logger.go b/logger/logger.go
new file mode 100644
index 0000000..8a7fe09
--- /dev/null
+++ b/logger/logger.go
@@ -0,0 +1,42 @@
+package logger
+
+import (
+	"os"
+
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+)
+
+const LOG_FILE_NAME = "harbourbridge.log"
+
+var Log *zap.Logger
+
+func InitializeLogger(inputLogLevel string) error {
+	// create zapper encoding config object
+	config := zap.NewProductionEncoderConfig()
+	// set logging timestamp format
+	config.EncodeTime = zapcore.ISO8601TimeEncoder
+	// create encoder for logs that are written to file
+	fileEncoder := zapcore.NewJSONEncoder(config)
+	// create encoder for logs that are written to console
+	// we create two encoders because we want to write human readable logs to console and
+	// JSON parsable logs to the file
+	consoleEncoder := zapcore.NewConsoleEncoder(config)
+	// specify log file.
+	logFile, _ := os.OpenFile(LOG_FILE_NAME, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+	writer := zapcore.AddSync(logFile)
+	// create and set the log level from the user input
+	zapLogLevel := new(zapcore.Level)
+	err := zapLogLevel.Set(inputLogLevel)
+	if err != nil {
+		return err
+	}
+	logLevel := zap.NewAtomicLevelAt(*zapLogLevel)
+	// create the logger
+	core := zapcore.NewTee(
+		zapcore.NewCore(fileEncoder, writer, logLevel),
+		zapcore.NewCore(consoleEncoder, zapcore.AddSync(os.Stdout), logLevel),
+	)
+	Log = zap.New(core, zap.AddCaller(), zap.AddStacktrace(zapcore.ErrorLevel))
+	return nil
+}
diff --git a/main.go b/main.go
index 76ae6d2..21ab899 100644
--- a/main.go
+++ b/main.go
@@ -35,6 +35,7 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/web"
 	"github.com/cloudspannerecosystem/harbourbridge/webv2"
 	"github.com/google/subcommands"
@@ -123,6 +124,11 @@ func main() {
 	fmt.Printf("\nWarning: Found usage of deprecated flags. Support for these " +
 		"flags will be discontinued soon.\nIt is recommended to use Harbourbridge " +
 		"using connection profiles. Checkout usage here: https://github.com/cloudspannerecosystem/harbourbridge/tree/master/cmd#command-line-flags\n\n")
+	err = logger.InitializeLogger("INFO")
+	if err != nil {
+		panic(fmt.Errorf("error initialising logger"))
+	}
+	defer logger.Log.Sync()
 	// Running HB CLI in global command line mode.
 	setupGlobalFlags()
 	flag.Usage = usage
diff --git a/sources/dynamodb/schema.go b/sources/dynamodb/schema.go
index 178b502..2b34256 100644
--- a/sources/dynamodb/schema.go
+++ b/sources/dynamodb/schema.go
@@ -20,6 +20,7 @@ import (
 	"log"
 	"math/big"
 	"sort"
+	"sync"
 
 	sp "cloud.google.com/go/spanner"
 	"github.com/aws/aws-sdk-go/aws"
@@ -214,8 +215,28 @@ func (isi InfoSchemaImpl) StartChangeDataCapture(ctx context.Context, conv *inte
 	return latestStreamArn, nil
 }
 
+// StartStreamingMigration starts the streaming migration process by creating a seperate
+// worker thread/goroutine for each table's DynamoDB Stream. It catches Ctrl+C signal if
+// customer wants to stop the process.
 func (isi InfoSchemaImpl) StartStreamingMigration(ctx context.Context, client *sp.Client, conv *internal.Conv, latestStreamArn map[string]interface{}) error {
-	// TODO(nareshz): work in progress
+	fmt.Println("Processing of DynamoDB Streams started...")
+	fmt.Println("Use Ctrl+C to stop the process.")
+
+	streamInfo := MakeStreamingInfo()
+	wg := &sync.WaitGroup{}
+
+	wg.Add(1)
+	go catchCtrlC(wg, streamInfo)
+
+	for srcTable, streamArn := range latestStreamArn {
+		streamInfo.makeRecordMaps(srcTable)
+
+		wg.Add(1)
+		go ProcessStream(wg, isi.DynamoStreamsClient, streamInfo, conv, streamArn.(string), srcTable)
+	}
+	wg.Wait()
+
+	fmt.Println("DynamoDB Streams processed successfully.")
 	return nil
 }
 
diff --git a/sources/dynamodb/streaming.go b/sources/dynamodb/streaming.go
index e2bb5eb..7031a18 100644
--- a/sources/dynamodb/streaming.go
+++ b/sources/dynamodb/streaming.go
@@ -15,10 +15,20 @@ package dynamodb
 
 import (
 	"fmt"
+	"os"
+	"os/signal"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/dynamodb"
 	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
+	"github.com/aws/aws-sdk-go/service/dynamodbstreams"
+	"github.com/aws/aws-sdk-go/service/dynamodbstreams/dynamodbstreamsiface"
+
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
 )
 
 // NewDynamoDBStream initializes a new DynamoDB Stream for a table with NEW_AND_OLD_IMAGES
@@ -58,3 +68,194 @@ func NewDynamoDBStream(client dynamodbiface.DynamoDBAPI, srcTable string) (strin
 		return *res.TableDescription.LatestStreamArn, nil
 	}
 }
+
+// catchCtrlC catches the Ctrl+C signal if customer wants to exit.
+func catchCtrlC(wg *sync.WaitGroup, streamInfo *StreamingInfo) {
+	defer wg.Done()
+	c := make(chan os.Signal)
+	signal.Notify(c, os.Interrupt, syscall.SIGTERM)
+	go func() {
+		<-c
+		streamInfo.UserExit = true
+	}()
+}
+
+// ProcessStream processes the latest enabled DynamoDB Stream for a table. It searches
+// for shards within stream and for each shard it creates a seperate working thread to
+// process records within it.
+func ProcessStream(wgStream *sync.WaitGroup, streamClient dynamodbstreamsiface.DynamoDBStreamsAPI, streamInfo *StreamingInfo, conv *internal.Conv, streamArn, srcTable string) {
+	defer wgStream.Done()
+	wgShard := &sync.WaitGroup{}
+
+	var lastProcessedShardId *string = nil
+	passAfterUserExit := false
+	for {
+		result, err := fetchShards(streamClient, lastProcessedShardId, streamArn)
+		if err != nil {
+			streamInfo.Unexpected(fmt.Sprintf("Couldn't fetch shards for table %s: %s", srcTable, err))
+			break
+		}
+		for _, shard := range result.Shards {
+			lastProcessedShardId = shard.ShardId
+
+			wgShard.Add(1)
+			go ProcessShard(wgShard, streamInfo, conv, streamClient, shard, streamArn, srcTable)
+		}
+
+		if result.LastEvaluatedShardId == nil && passAfterUserExit {
+			break
+		}
+		if streamInfo.UserExit {
+			passAfterUserExit = true
+		} else if len(result.Shards) == 0 {
+			time.Sleep(10 * time.Second)
+		}
+	}
+	wgShard.Wait()
+}
+
+// fetchShards fetches latest unprocessed shards from a given DynamoDB Stream after the lastProcessedShardId.
+func fetchShards(streamClient dynamodbstreamsiface.DynamoDBStreamsAPI, lastProcessedShardId *string, streamArn string) (*dynamodbstreams.StreamDescription, error) {
+	describeStreamInput := &dynamodbstreams.DescribeStreamInput{
+		ExclusiveStartShardId: lastProcessedShardId,
+		StreamArn:             &streamArn,
+	}
+	result, err := streamClient.DescribeStream(describeStreamInput)
+	if err != nil {
+		return nil, fmt.Errorf("unexpected call to DescribeStream: %v", err)
+	}
+	return result.StreamDescription, nil
+}
+
+// checkTrimmedDataError checks if the error is an TrimmedDataAccessException.
+func checkTrimmedDataError(err error) bool {
+	return strings.Contains(err.Error(), "TrimmedDataAccessException")
+}
+
+// ProcessShard processes records within a shard starting from the first unexpired record. It
+// doesn't start processing unless parent shard is processed. For closed shards this process is
+// completed after processing all records but for open shards it keeps searching for new records
+// until shards gets closed or customer calls for a exit.
+func ProcessShard(wgShard *sync.WaitGroup, streamInfo *StreamingInfo, conv *internal.Conv, streamClient dynamodbstreamsiface.DynamoDBStreamsAPI, shard *dynamodbstreams.Shard, streamArn, srcTable string) {
+	defer wgShard.Done()
+
+	waitForParentShard(streamInfo, shard.ParentShardId)
+
+	shardId := *shard.ShardId
+	streamInfo.SetShardStatus(shardId, false)
+
+	var lastEvaluatedSequenceNumber *string = nil
+	passAfterUserExit := false
+	retryCount := 0
+	for {
+		shardIterator, err := getShardIterator(streamClient, lastEvaluatedSequenceNumber, shardId, streamArn)
+		if err != nil {
+			if checkTrimmedDataError(err) {
+				lastEvaluatedSequenceNumber = nil
+				continue
+			} else {
+				streamInfo.Unexpected(fmt.Sprintf("Couldn't get shardIterator for table %s: %s", srcTable, err))
+				break
+			}
+		}
+
+		getRecordsOutput, err := getRecords(streamClient, shardIterator)
+		if err != nil {
+			// In case of closed shards, after all data records get expired it still returns a non-nil
+			// shardIterator for GetShardIterator query. Using this shardIterator for GetRecords
+			// API call results in TrimmedDataAccessException. This will result in same steps being
+			// followed again and again. To handle this a retry limit of 5 is set.
+			if checkTrimmedDataError(err) && retryCount < 5 {
+				lastEvaluatedSequenceNumber = nil
+				retryCount++
+				continue
+			} else {
+				streamInfo.Unexpected(fmt.Sprintf("Couldn't fetch records for table %s: %s", srcTable, err))
+				break
+			}
+		} else {
+			retryCount = 0
+		}
+
+		records := getRecordsOutput.Records
+		for _, record := range records {
+			ProcessRecord(conv, streamInfo, record, srcTable)
+			lastEvaluatedSequenceNumber = record.Dynamodb.SequenceNumber
+		}
+
+		if getRecordsOutput.NextShardIterator == nil || passAfterUserExit {
+			break
+		}
+		if streamInfo.UserExit {
+			passAfterUserExit = true
+		} else if len(records) == 0 {
+			time.Sleep(5 * time.Second)
+		}
+	}
+	streamInfo.SetShardStatus(shardId, true)
+}
+
+// waitForParentShard checks every 6 seconds if parentShard is processed or
+// not and waits as long as parent shard is not processed.
+func waitForParentShard(streamInfo *StreamingInfo, parentShard *string) {
+	if parentShard != nil {
+		for {
+			streamInfo.lock.Lock()
+			done := streamInfo.ShardProcessed[*parentShard]
+			streamInfo.lock.Unlock()
+			if done {
+				return
+			} else {
+				time.Sleep(6 * time.Second)
+			}
+		}
+	}
+}
+
+// getShardIterator returns an iterator to find records based on the lastEvaluatedSequence number.
+// If lastEvaluatedSequenceNumber is nil then it uses TrimHorizon as shardIterator type to point to first
+// non-expired record otherwise it finds the first unprocessed record after lastEvaluatedSequence number using
+// AfterSequenceNumber shardIterator type.
+func getShardIterator(streamClient dynamodbstreamsiface.DynamoDBStreamsAPI, lastEvaluatedSequenceNumber *string, shardId, streamArn string) (*string, error) {
+	var getShardIteratorInput *dynamodbstreams.GetShardIteratorInput
+	if lastEvaluatedSequenceNumber == nil {
+		getShardIteratorInput = &dynamodbstreams.GetShardIteratorInput{
+			ShardId:           &shardId,
+			ShardIteratorType: aws.String(dynamodbstreams.ShardIteratorTypeTrimHorizon),
+			StreamArn:         &streamArn,
+		}
+	} else {
+		getShardIteratorInput = &dynamodbstreams.GetShardIteratorInput{
+			SequenceNumber:    lastEvaluatedSequenceNumber,
+			ShardId:           &shardId,
+			ShardIteratorType: aws.String(dynamodbstreams.ShardIteratorTypeAfterSequenceNumber),
+			StreamArn:         &streamArn,
+		}
+	}
+	result, err := streamClient.GetShardIterator(getShardIteratorInput)
+	if err != nil {
+		err = fmt.Errorf("unexpected call to GetShardIterator: %v", err)
+		return nil, err
+	}
+	return result.ShardIterator, nil
+}
+
+// getRecords fetches the records from DynamoDB Streams by using the shardIterator.
+func getRecords(streamClient dynamodbstreamsiface.DynamoDBStreamsAPI, shardIterator *string) (*dynamodbstreams.GetRecordsOutput, error) {
+	getRecordsInput := &dynamodbstreams.GetRecordsInput{
+		ShardIterator: shardIterator,
+	}
+	result, err := streamClient.GetRecords(getRecordsInput)
+	if err != nil {
+		err = fmt.Errorf("unexpected call to GetRecords: %v", err)
+		return result, err
+	}
+	return result, nil
+}
+
+// ProcessRecord processes records retrieved from shards.
+func ProcessRecord(conv *internal.Conv, streamInfo *StreamingInfo, record *dynamodbstreams.Record, srcTable string) {
+	streamInfo.StatsAddRecord(srcTable, *record.EventName)
+	// TODO(nareshz): work in progress
+	streamInfo.StatsAddRecordProcessed()
+}
diff --git a/sources/dynamodb/streaming_info.go b/sources/dynamodb/streaming_info.go
new file mode 100644
index 0000000..95cfd81
--- /dev/null
+++ b/sources/dynamodb/streaming_info.go
@@ -0,0 +1,90 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package dynamodb
+
+import (
+	"sync"
+
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+)
+
+// StreamingInfo contains information related to processing of DynamoDB Streams.
+type StreamingInfo struct {
+	Records          map[string]map[string]int64 // Tablewise count of records received from DynamoDB Streams, broken down by record type i.e. INSERT, MODIFY & REMOVE.
+	recordsProcessed int64                       // Count of total records processed to Cloud Spanner(includes records which generated error as well).
+	ShardProcessed   map[string]bool             // Processing status of a shard, (default false i.e. unprocessed).
+	UserExit         bool                        // flag confirming if customer wants to exit or not, (false until user presses Ctrl+C).
+	Unexpecteds      map[string]int64            // Count of unexpected conditions, broken down by condition description.
+	lock             sync.Mutex
+}
+
+func MakeStreamingInfo() *StreamingInfo {
+	return &StreamingInfo{
+		Records:          make(map[string]map[string]int64),
+		recordsProcessed: int64(0),
+		ShardProcessed:   make(map[string]bool),
+		Unexpecteds:      make(map[string]int64),
+		UserExit:         false,
+		lock:             sync.Mutex{},
+	}
+}
+
+// makeRecordMaps initializes maps used to stores record count for
+// a given table.
+func (info *StreamingInfo) makeRecordMaps(srcTable string) {
+	info.Records[srcTable] = make(map[string]int64)
+}
+
+// SetShardStatus changes the processing status of a shard.
+//
+// true -> shard processed and vice versa.
+func (info *StreamingInfo) SetShardStatus(shardId string, status bool) {
+	info.lock.Lock()
+	info.ShardProcessed[shardId] = status
+	info.lock.Unlock()
+}
+
+// StatsAddRecord increases the count of records read from DynamoDB Streams
+// based on the table name and record type.
+func (info *StreamingInfo) StatsAddRecord(srcTable, recordType string) {
+	info.lock.Lock()
+	info.Records[srcTable][recordType]++
+	info.lock.Unlock()
+}
+
+// StatsAddRecordProcessed increases the count of total records processed to Cloud Spanner.
+func (info *StreamingInfo) StatsAddRecordProcessed() {
+	info.lock.Lock()
+	info.recordsProcessed++
+	info.lock.Unlock()
+}
+
+// Unexpected records stats about corner-cases and conditions
+// that were not expected.
+func (info *StreamingInfo) Unexpected(u string) {
+	info.lock.Lock()
+	internal.VerbosePrintf("Unexpected condition: %s\n", u)
+	// Limit size of unexpected map. If over limit, then only
+	// update existing entries.
+	if _, ok := info.Unexpecteds[u]; ok || len(info.Unexpecteds) < 1000 {
+		info.Unexpecteds[u]++
+	}
+	info.lock.Unlock()
+}
+
+// TotalUnexpecteds returns the total number of distinct unexpected conditions
+// encountered during processing of DynamoDB Streams.
+func (info *StreamingInfo) TotalUnexpecteds() int64 {
+	return int64(len(info.Unexpecteds))
+}
diff --git a/sources/mysql/mysqldump.go b/sources/mysql/mysqldump.go
index 95a65aa..95e7ca5 100644
--- a/sources/mysql/mysqldump.go
+++ b/sources/mysql/mysqldump.go
@@ -22,6 +22,7 @@ import (
 	"strings"
 
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/pingcap/tidb/parser"
@@ -76,6 +77,7 @@ func processMySQLDump(conv *internal.Conv, r *internal.Reader) error {
 		for _, stmt := range stmts {
 			isInsert := processStatement(conv, stmt)
 			internal.VerbosePrintf("Parsed SQL command at line=%d/fpos=%d: %d stmts (%d lines, %d bytes) Insert Statement=%v\n", startLine, startOffset, 1, r.LineNumber-startLine, len(b), isInsert)
+			logger.Log.Debug(fmt.Sprintf("Parsed SQL command at line=%d/fpos=%d: %d stmts (%d lines, %d bytes) Insert Statement=%v\n", startLine, startOffset, 1, r.LineNumber-startLine, len(b), isInsert))
 		}
 		if r.EOF {
 			break
@@ -227,6 +229,8 @@ func processCreateTable(conv *internal.Conv, stmt *ast.CreateTableStmt) {
 	}
 	tableName, err := getTableName(stmt.Table)
 	internal.VerbosePrintf("processing create table elem=%s stmt=%v\n", tableName, stmt)
+	logger.Log.Debug(fmt.Sprintf("processing create table elem=%s stmt=%v\n", tableName, stmt))
+
 	if err != nil {
 		logStmtError(conv, stmt, fmt.Errorf("can't get table name: %w", err))
 		return
@@ -574,6 +578,8 @@ func handleParseError(conv *internal.Conv, chunk string, err error, l [][]byte)
 			if conv.SchemaMode() {
 				conv.Unexpected(fmt.Sprintf("Unsupported datatype '%s' encountered while parsing following statement at line number %d : \n%s", spatial, len(l), chunk))
 				internal.VerbosePrintf("Converting datatype '%s' to 'Text' and retrying to parse the statement\n", spatial)
+				logger.Log.Debug(fmt.Sprintf("Converting datatype '%s' to 'Text' and retrying to parse the statement\n", spatial))
+
 			}
 			return handleSpatialDatatype(conv, chunk, l)
 		}
diff --git a/sources/postgres/pgdump.go b/sources/postgres/pgdump.go
index f64cc1f..28c98ab 100644
--- a/sources/postgres/pgdump.go
+++ b/sources/postgres/pgdump.go
@@ -24,6 +24,7 @@ import (
 	pg_query "github.com/pganalyze/pg_query_go/v2"
 
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 )
@@ -70,6 +71,7 @@ func processPgDump(conv *internal.Conv, r *internal.Reader) error {
 		}
 		ci := processStatements(conv, stmts)
 		internal.VerbosePrintf("Parsed SQL command at line=%d/fpos=%d: %d stmts (%d lines, %d bytes) ci=%v\n", startLine, startOffset, len(stmts), r.LineNumber-startLine, len(b), ci != nil)
+		logger.Log.Debug(fmt.Sprintf("Parsed SQL command at line=%d/fpos=%d: %d stmts (%d lines, %d bytes) ci=%v\n", startLine, startOffset, len(stmts), r.LineNumber-startLine, len(b), ci != nil))
 		if ci != nil {
 			switch ci.stmt {
 			case copyFrom:
@@ -131,10 +133,12 @@ func readAndParseChunk(conv *internal.Conv, r *internal.Reader) ([]byte, []*pg_q
 
 func processCopyBlock(conv *internal.Conv, srcTable string, srcCols []string, r *internal.Reader) {
 	internal.VerbosePrintf("Parsing COPY-FROM stdin block starting at line=%d/fpos=%d\n", r.LineNumber, r.Offset)
+	logger.Log.Debug(fmt.Sprintf("Parsing COPY-FROM stdin block starting at line=%d/fpos=%d\n", r.LineNumber, r.Offset))
 	for {
 		b := r.ReadLine()
 		if string(b) == "\\.\n" || string(b) == "\\.\r\n" {
 			internal.VerbosePrintf("Parsed COPY-FROM stdin block ending at line=%d/fpos=%d\n", r.LineNumber, r.Offset)
+			logger.Log.Debug(fmt.Sprintf("Parsed COPY-FROM stdin block ending at line=%d/fpos=%d\n", r.LineNumber, r.Offset))
 			return
 		}
 		if r.EOF {
@@ -269,6 +273,7 @@ func processAlterTableStmt(conv *internal.Conv, n *pg_query.AlterTableStmt) {
 		// in verbose mode, but otherwise  we just skip these statements.
 		conv.SkipStatement(printNodeType(n))
 		internal.VerbosePrintf("Processing %v statement: table %s not found", printNodeType(n), table)
+		logger.Log.Debug(fmt.Sprintf("Processing %v statement: table %s not found", printNodeType(n), table))
 	}
 }
 
@@ -289,6 +294,8 @@ func processCreateStmt(conv *internal.Conv, n *pg_query.CreateStmt) {
 		conv.SkipStatement(printNodeType(n))
 		conv.Unexpected(fmt.Sprintf("Found inherited table %s -- we do not currently handle inherited tables", table))
 		internal.VerbosePrintf("Processing %v statement: table %s is inherited table", printNodeType(n), table)
+		logger.Log.Debug(fmt.Sprintf("Processing %v statement: table %s is inherited table", printNodeType(n), table))
+
 		return
 	}
 	var constraints []constraint
@@ -355,6 +362,8 @@ func processInsertStmt(conv *internal.Conv, n *pg_query.InsertStmt) *copyOrInser
 		// for a table is that it is an inherited table - we skip all inherited tables.
 		conv.SkipStatement(printNodeType(n))
 		internal.VerbosePrintf("Processing %v statement: table %s not found", printNodeType(n), table)
+		logger.Log.Debug(fmt.Sprintf("Processing %v statement: table %s is inherited table", printNodeType(n), table))
+
 		return nil
 	}
 	conv.StatsAddRow(table, conv.SchemaMode())
@@ -399,6 +408,7 @@ func processCopyStmt(conv *internal.Conv, n *pg_query.CopyStmt) *copyOrInsert {
 		// for a table is that it is an inherited table - we skip all inherited tables.
 		conv.SkipStatement(printNodeType(n))
 		internal.VerbosePrintf("Processing %v statement: table %s not found", printNodeType(n), table)
+		logger.Log.Debug(fmt.Sprintf("Processing %v statement: table %s is inherited table", printNodeType(n), table))
 		return &copyOrInsert{stmt: copyFrom, table: table, cols: []string{}}
 	}
 	var cols []string
diff --git a/spanner/writer/batchwriter.go b/spanner/writer/batchwriter.go
index c82a78b..2ad06fd 100644
--- a/spanner/writer/batchwriter.go
+++ b/spanner/writer/batchwriter.go
@@ -25,6 +25,7 @@ import (
 	"unsafe"
 
 	sp "cloud.google.com/go/spanner"
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 )
 
 // Parameters used to control building batches to write to Spanner.
@@ -131,6 +132,9 @@ func (bw *BatchWriter) Flush() {
 				fmt.Printf("Starting write of %d rows to Spanner (%d bytes, %d mutations) [%d in progress]\n",
 					len(m), bytes, count, atomic.LoadInt64(&bw.async.writes))
 			}
+			logger.Log.Debug(fmt.Sprintf("Starting write of %d rows to Spanner (%d bytes, %d mutations) [%d in progress]\n",
+				len(m), bytes, count, atomic.LoadInt64(&bw.async.writes)))
+
 			bw.startWrite(m)
 		} else {
 			time.Sleep(10 * time.Millisecond)
@@ -218,7 +222,7 @@ func (bw *BatchWriter) errorStats(rows []*row, err error, retry bool) {
 	if bw.verbose {
 		fmt.Printf("Error while writing %d rows to Spanner: %v\n", len(rows), err)
 	}
-
+	logger.Log.Debug(fmt.Sprintf("Error while writing %d rows to Spanner: %v\n", len(rows), err))
 	bw.async.lock.Lock()
 	defer bw.async.lock.Unlock()
 
@@ -258,6 +262,9 @@ func (bw *BatchWriter) doWriteAndHandleErrors(rows []*row) {
 			if hitRetryLimit && bw.verbose {
 				fmt.Printf("Have hit %d retries: will not do any more\n", atomic.LoadInt64(&bw.async.retries))
 			}
+			if hitRetryLimit {
+				logger.Log.Debug(fmt.Sprintf("Have hit %d retries: will not do any more\n", atomic.LoadInt64(&bw.async.retries)))
+			}
 			return
 		}
 		// Split into 10 pieces and retry. This is useful
@@ -306,6 +313,8 @@ func (bw *BatchWriter) writeData() {
 				fmt.Printf("Starting write of %d rows to Spanner (%d bytes, %d mutations) [%d in progress]\n",
 					len(m), bytes, count, atomic.LoadInt64(&bw.async.writes))
 			}
+			logger.Log.Debug(fmt.Sprintf("Starting write of %d rows to Spanner (%d bytes, %d mutations) [%d in progress]\n",
+				len(m), bytes, count, atomic.LoadInt64(&bw.async.writes)))
 			bw.startWrite(m)
 		} else {
 			if bw.rBytes < bw.bytesLimit {


Test patch:
diff --git a/internal/convert_test.go b/internal/convert_test.go
index 82832ab..24b76c5 100644
--- a/internal/convert_test.go
+++ b/internal/convert_test.go
@@ -18,14 +18,18 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
 
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
 // This file contains very basic tests of Conv API functionality.
 // Most of the Conv APIs are also tested in process_test.go (where
 // they are tested using data from schema/data conversion).
-
+func init() {
+	logger.Log = zap.NewNop()
+}
 func TestSetSchemaMode(t *testing.T) {
 	conv := MakeConv()
 	conv.SetSchemaMode()
diff --git a/sources/common/utils_test.go b/sources/common/utils_test.go
index c259b8b..8937b15 100644
--- a/sources/common/utils_test.go
+++ b/sources/common/utils_test.go
@@ -18,9 +18,15 @@ import (
 	"testing"
 
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
 )
 
+func init() {
+	logger.Log = zap.NewNop()
+}
+
 func TestToNotNull(t *testing.T) {
 	conv := internal.MakeConv()
 	assert.Equal(t, false, ToNotNull(conv, "YES"))
diff --git a/sources/dynamodb/schema_test.go b/sources/dynamodb/schema_test.go
index a41276b..a5ee813 100644
--- a/sources/dynamodb/schema_test.go
+++ b/sources/dynamodb/schema_test.go
@@ -26,13 +26,19 @@ import (
 	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
 	"github.com/aws/aws-sdk-go/service/dynamodbstreams"
 	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
 
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
+func init() {
+	logger.Log = zap.NewNop()
+}
+
 type mockDynamoClient struct {
 	listTableCallCount     int
 	listTableOutputs       []dynamodb.ListTablesOutput
diff --git a/sources/dynamodb/streaming_info_test.go b/sources/dynamodb/streaming_info_test.go
new file mode 100644
index 0000000..d01b295
--- /dev/null
+++ b/sources/dynamodb/streaming_info_test.go
@@ -0,0 +1,74 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package dynamodb
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestInfo_TotalUnexpecteds(t *testing.T) {
+	streamInfo := MakeStreamingInfo()
+	streamInfo.Unexpected("testing-unexpecteds-faced-1")
+	streamInfo.Unexpected("testing-unexpecteds-faced-2")
+	assert.Equal(t, int64(2), streamInfo.TotalUnexpecteds())
+}
+
+func TestStreamingInfo_SetShardStatus(t *testing.T) {
+	streamInfo := MakeStreamingInfo()
+	shardId := "testShardId"
+	streamInfo.SetShardStatus(shardId, true)
+	assert.Equal(t, true, streamInfo.ShardProcessed[shardId])
+}
+
+func sumNestedMapValues(mp map[string]map[string]int64) int64 {
+	n := int64(0)
+	for _, x := range mp {
+		for _, y := range x {
+			n += y
+		}
+	}
+	return n
+}
+
+func TestStreamingInfo_StatsAddRecord(t *testing.T) {
+	streamInfo := MakeStreamingInfo()
+
+	streamInfo.makeRecordMaps("testtable1")
+	streamInfo.makeRecordMaps("testtable2")
+	assert.NotNil(t, streamInfo.Records["testtable1"])
+
+	streamInfo.StatsAddRecord("testtable1", "INSERT")
+	streamInfo.StatsAddRecord("testtable2", "REMOVE")
+
+	assert.Equal(t, int64(2), sumNestedMapValues(streamInfo.Records))
+}
+
+func TestStreamingInfo_StatsAddRecordProcessed(t *testing.T) {
+	streamInfo := MakeStreamingInfo()
+	for i := 0; i < 3; i++ {
+		streamInfo.StatsAddRecordProcessed()
+	}
+	assert.Equal(t, int64(3), streamInfo.recordsProcessed)
+}
+
+func TestStreamingInfo_makeRecordMaps(t *testing.T) {
+	streamInfo := MakeStreamingInfo()
+	table := "testtable"
+
+	assert.Nil(t, streamInfo.Records[table])
+	streamInfo.makeRecordMaps(table)
+	assert.NotNil(t, streamInfo.Records[table])
+}
diff --git a/sources/dynamodb/streaming_test.go b/sources/dynamodb/streaming_test.go
new file mode 100644
index 0000000..6d88240
--- /dev/null
+++ b/sources/dynamodb/streaming_test.go
@@ -0,0 +1,300 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package dynamodb
+
+import (
+	"fmt"
+	"reflect"
+	"sync"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/dynamodbstreams"
+	"github.com/aws/aws-sdk-go/service/dynamodbstreams/dynamodbstreamsiface"
+	"github.com/stretchr/testify/assert"
+)
+
+type mockDynamoStreamsClient struct {
+	describeStreamOutputs                []dynamodbstreams.DescribeStreamOutput
+	describeStreamCallCount              int
+	listStreamsOutputs                   []dynamodbstreams.ListStreamsOutput
+	listStreamsCallCount                 int
+	getShardIteratorOutputsTrimHorizon   []dynamodbstreams.GetShardIteratorOutput
+	getShardIteratorCallCountTrimHorizon int
+	getShardIteratorOutputsSeqNum        []dynamodbstreams.GetShardIteratorOutput
+	getShardIteratorCallCountSeqNum      int
+	getRecordsOutputs                    []dynamodbstreams.GetRecordsOutput
+	getRecordsCallCount                  int
+	dynamodbstreamsiface.DynamoDBStreamsAPI
+}
+
+func (m *mockDynamoStreamsClient) DescribeStream(input *dynamodbstreams.DescribeStreamInput) (*dynamodbstreams.DescribeStreamOutput, error) {
+	if m.describeStreamCallCount >= len(m.describeStreamOutputs) {
+		return nil, fmt.Errorf("unexpected call to DescribeStream: %v", input)
+	}
+	m.describeStreamCallCount++
+	return &m.describeStreamOutputs[m.describeStreamCallCount-1], nil
+}
+
+func (m *mockDynamoStreamsClient) ListStreams(input *dynamodbstreams.ListStreamsInput) (*dynamodbstreams.ListStreamsOutput, error) {
+	if m.listStreamsCallCount >= len(m.listStreamsOutputs) {
+		return nil, fmt.Errorf("unexpected call to ListStreams: %v", input)
+	}
+	m.listStreamsCallCount++
+	return &m.listStreamsOutputs[m.listStreamsCallCount-1], nil
+}
+
+func (m *mockDynamoStreamsClient) GetShardIterator(input *dynamodbstreams.GetShardIteratorInput) (*dynamodbstreams.GetShardIteratorOutput, error) {
+	if *input.ShardIteratorType == "TRIM_HORIZON" {
+		if m.getShardIteratorCallCountTrimHorizon >= len(m.getShardIteratorOutputsTrimHorizon) {
+			return nil, fmt.Errorf("unexpected call to GetShardIterator: %v", input)
+		}
+		m.getShardIteratorCallCountTrimHorizon++
+		return &m.getShardIteratorOutputsTrimHorizon[m.getShardIteratorCallCountTrimHorizon-1], nil
+	} else {
+		if m.getShardIteratorCallCountSeqNum >= len(m.getShardIteratorOutputsSeqNum) {
+			return nil, fmt.Errorf("unexpected call to GetShardIterator: %v", input)
+		}
+		m.getShardIteratorCallCountSeqNum++
+		return &m.getShardIteratorOutputsSeqNum[m.getShardIteratorCallCountSeqNum-1], nil
+	}
+}
+
+func (m *mockDynamoStreamsClient) GetRecords(input *dynamodbstreams.GetRecordsInput) (*dynamodbstreams.GetRecordsOutput, error) {
+	if m.getRecordsCallCount >= len(m.getRecordsOutputs) {
+		return nil, fmt.Errorf("unexpected call to GetShardIterator: %v", input)
+	}
+	m.getRecordsCallCount++
+	return &m.getRecordsOutputs[m.getRecordsCallCount-1], nil
+}
+
+func TestProcessStream(t *testing.T) {
+	streamInfo := MakeStreamingInfo()
+	streamInfo.UserExit = true
+	wgStream := &sync.WaitGroup{}
+
+	streamArn := [2]string{"streamArn1", "streamArn2"}
+	srcTableName := [2]string{"table1", "table2"}
+
+	describeStreamOutputs := []dynamodbstreams.DescribeStreamOutput{
+		{
+			StreamDescription: &dynamodbstreams.StreamDescription{
+				LastEvaluatedShardId: nil,
+				Shards:               []*dynamodbstreams.Shard{},
+				StreamArn:            &streamArn[0],
+			},
+		},
+		{
+			StreamDescription: &dynamodbstreams.StreamDescription{
+				LastEvaluatedShardId: nil,
+				Shards:               []*dynamodbstreams.Shard{},
+				StreamArn:            &streamArn[0],
+			},
+		},
+		{
+			StreamDescription: &dynamodbstreams.StreamDescription{
+				LastEvaluatedShardId: nil,
+				Shards:               []*dynamodbstreams.Shard{},
+				StreamArn:            &streamArn[1],
+			},
+		},
+	}
+	streamsClient := &mockDynamoStreamsClient{
+		describeStreamOutputs: describeStreamOutputs,
+	}
+
+	wgStream.Add(2)
+	for i := 0; i < 2; i++ {
+		ProcessStream(wgStream, streamsClient, streamInfo, nil, streamArn[i], srcTableName[i])
+	}
+	wgStream.Wait()
+	assert.Equal(t, int64(1), streamInfo.TotalUnexpecteds())
+}
+
+func Test_fetchShards(t *testing.T) {
+	streamArn := "testStreamArn"
+	tableName := "testTable"
+	type args struct {
+		streamClient         dynamodbstreamsiface.DynamoDBStreamsAPI
+		lastEvaluatedShardId *string
+		streamArn            string
+	}
+	streamDescription := dynamodbstreams.StreamDescription{
+		LastEvaluatedShardId: aws.String("shard2"),
+		Shards: []*dynamodbstreams.Shard{
+			{
+				ShardId: aws.String("shard1"),
+			},
+			{
+				ShardId: aws.String("shard2"),
+			},
+		},
+		StreamArn: &streamArn,
+		TableName: &tableName,
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    *dynamodbstreams.StreamDescription
+		wantErr bool
+	}{
+		{
+			name: "test for correctness of output",
+			args: args{
+				streamClient: &mockDynamoStreamsClient{
+					describeStreamOutputs: []dynamodbstreams.DescribeStreamOutput{
+						{
+							StreamDescription: &streamDescription,
+						},
+					},
+				},
+				streamArn: streamArn,
+			},
+			want:    &streamDescription,
+			wantErr: false,
+		},
+		{
+			name: "test for checking api failures",
+			args: args{
+				streamClient: &mockDynamoStreamsClient{},
+				streamArn:    streamArn,
+			},
+			want:    nil,
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := fetchShards(tt.args.streamClient, tt.args.lastEvaluatedShardId, tt.args.streamArn)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("fetchShards() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("fetchShards() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_getShardIterator(t *testing.T) {
+	shardIteratorTrimHorizon := "testShardIteratorTrimHorizon"
+	shardIteratorSeqNum := "testShardIteratorSeqNum"
+	lastEvaluatedSequenceNumber := "0000000000000001"
+
+	mockStreamsClient := &mockDynamoStreamsClient{
+		getShardIteratorOutputsTrimHorizon: []dynamodbstreams.GetShardIteratorOutput{
+			{
+				ShardIterator: &shardIteratorTrimHorizon,
+			},
+		},
+		getShardIteratorOutputsSeqNum: []dynamodbstreams.GetShardIteratorOutput{
+			{
+				ShardIterator: &shardIteratorSeqNum,
+			},
+		},
+	}
+
+	shardIterator, err := getShardIterator(mockStreamsClient, nil, "", "")
+	assert.Nil(t, err)
+	assert.Equal(t, shardIteratorTrimHorizon, *shardIterator)
+
+	shardIterator, err = getShardIterator(mockStreamsClient, &lastEvaluatedSequenceNumber, "", "")
+	assert.Nil(t, err)
+	assert.Equal(t, shardIteratorSeqNum, *shardIterator)
+
+	_, err = getShardIterator(mockStreamsClient, nil, "", "")
+	assert.NotNil(t, err)
+}
+
+func Test_getRecords(t *testing.T) {
+	mockStreamsClient := &mockDynamoStreamsClient{
+		getRecordsOutputs: []dynamodbstreams.GetRecordsOutput{
+			{
+				NextShardIterator: nil,
+				Records: []*dynamodbstreams.Record{
+					{
+						EventName: aws.String("INSERT"),
+					},
+					{
+						EventName: aws.String("REMOVE"),
+					},
+					{
+						EventName: aws.String("MODIFY"),
+					},
+					{
+						EventName: aws.String("INSERT"),
+					},
+				},
+			},
+		},
+	}
+	shardIterator := "testShardIterator"
+	result, err := getRecords(mockStreamsClient, &shardIterator)
+	assert.Nil(t, err)
+	assert.Equal(t, int(4), len(result.Records))
+	mp := make(map[string]int)
+	for i := 0; i < len(result.Records); i++ {
+		mp[*result.Records[i].EventName]++
+	}
+	assert.Equal(t, int(1), mp["REMOVE"])
+	assert.Equal(t, int(2), mp["INSERT"])
+	assert.Equal(t, int(1), mp["MODIFY"])
+
+	_, err = getRecords(mockStreamsClient, &shardIterator)
+	assert.NotNil(t, err)
+}
+
+func TestProcessShard(t *testing.T) {
+	wgShard := &sync.WaitGroup{}
+	streamInfo := MakeStreamingInfo()
+	streamInfo.UserExit = true
+	shardIterator_TrimHorizon := "testShardIteratorTrimHorizon"
+
+	mockStreamClient := &mockDynamoStreamsClient{
+		getShardIteratorOutputsTrimHorizon: []dynamodbstreams.GetShardIteratorOutput{
+			{
+				ShardIterator: &shardIterator_TrimHorizon,
+			},
+			{
+				ShardIterator: &shardIterator_TrimHorizon,
+			},
+		},
+		getRecordsOutputs: []dynamodbstreams.GetRecordsOutput{
+			{
+				NextShardIterator: nil,
+				Records:           []*dynamodbstreams.Record{},
+			},
+		},
+	}
+	shardId := "testShardId"
+	shard := &dynamodbstreams.Shard{
+		SequenceNumberRange: &dynamodbstreams.SequenceNumberRange{
+			EndingSequenceNumber:   aws.String("10"),
+			StartingSequenceNumber: aws.String("10"),
+		},
+		ShardId: &shardId,
+	}
+	streamArn := "testStreamArn"
+	srcTable := "testSrcTable"
+
+	wgShard.Add(1)
+	ProcessShard(wgShard, streamInfo, nil, mockStreamClient, shard, streamArn, srcTable)
+	assert.Equal(t, true, streamInfo.ShardProcessed[*shard.ShardId])
+
+	wgShard.Add(1)
+	ProcessShard(wgShard, streamInfo, nil, mockStreamClient, shard, streamArn, srcTable)
+	assert.Equal(t, int64(1), streamInfo.TotalUnexpecteds())
+	assert.Equal(t, true, streamInfo.ShardProcessed[*shard.ShardId])
+}
diff --git a/sources/mysql/data_test.go b/sources/mysql/data_test.go
index c48baae..a736bc9 100644
--- a/sources/mysql/data_test.go
+++ b/sources/mysql/data_test.go
@@ -23,12 +23,18 @@ import (
 	"cloud.google.com/go/civil"
 	"cloud.google.com/go/spanner"
 	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
 
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
+func init() {
+	logger.Log = zap.NewNop()
+}
+
 type spannerData struct {
 	table string
 	cols  []string
diff --git a/sources/oracle/toddl_test.go b/sources/oracle/toddl_test.go
index 137693a..d2e8e43 100644
--- a/sources/oracle/toddl_test.go
+++ b/sources/oracle/toddl_test.go
@@ -19,12 +19,18 @@ import (
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
 )
 
+func init() {
+	logger.Log = zap.NewNop()
+}
+
 func TestToSpannerType(t *testing.T) {
 	conv := internal.MakeConv()
 	conv.SetSchemaMode()
diff --git a/sources/postgres/infoschema_test.go b/sources/postgres/infoschema_test.go
index fe61a5c..66459f6 100644
--- a/sources/postgres/infoschema_test.go
+++ b/sources/postgres/infoschema_test.go
@@ -24,12 +24,18 @@ import (
 	"cloud.google.com/go/spanner"
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
 )
 
+func init() {
+	logger.Log = zap.NewNop()
+}
+
 type mockSpec struct {
 	query string
 	args  []driver.Value   // Query args.
diff --git a/sources/sqlserver/infoschema_test.go b/sources/sqlserver/infoschema_test.go
index e392497..fb2bc74 100644
--- a/sources/sqlserver/infoschema_test.go
+++ b/sources/sqlserver/infoschema_test.go
@@ -21,11 +21,17 @@ import (
 
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
 )
 
+func init() {
+	logger.Log = zap.NewNop()
+}
+
 type mockSpec struct {
 	query string
 	args  []driver.Value   // Query args.
diff --git a/spanner/writer/batchwriter_test.go b/spanner/writer/batchwriter_test.go
index 4f8a503..aba8d12 100644
--- a/spanner/writer/batchwriter_test.go
+++ b/spanner/writer/batchwriter_test.go
@@ -25,9 +25,15 @@ import (
 	"time"
 
 	sp "cloud.google.com/go/spanner"
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
 )
 
+func init() {
+	logger.Log = zap.NewNop()
+}
+
 // TestFlush tests NewBatchWriter, AddRow and Flush.
 func TestFlush(t *testing.T) {
 	tests := []struct {
diff --git a/testing/conversion/conversion_test.go b/testing/conversion/conversion_test.go
index f3c0c57..bf1fb46 100644
--- a/testing/conversion/conversion_test.go
+++ b/testing/conversion/conversion_test.go
@@ -30,8 +30,10 @@ import (
 
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
 
 	database "cloud.google.com/go/spanner/admin/database/apiv1"
 	databasepb "google.golang.org/genproto/googleapis/spanner/admin/database/v1"
@@ -45,6 +47,10 @@ var (
 	databaseAdmin *database.DatabaseAdminClient
 )
 
+func init() {
+	logger.Log = zap.NewNop()
+}
+
 func TestMain(m *testing.M) {
 	cleanup := initTests()
 	res := m.Run()
diff --git a/testing/mysql/integration_test.go b/testing/mysql/integration_test.go
index 1c1539a..c090dd7 100644
--- a/testing/mysql/integration_test.go
+++ b/testing/mysql/integration_test.go
@@ -29,8 +29,10 @@ import (
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
+	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/testing/common"
 	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
 
 	"cloud.google.com/go/spanner"
 	database "cloud.google.com/go/spanner/admin/database/apiv1"
@@ -39,6 +41,10 @@ import (
 	databasepb "google.golang.org/genproto/googleapis/spanner/admin/database/v1"
 )
 
+func init() {
+	logger.Log = zap.NewNop()
+}
+
 var (
 	projectID  string
 	instanceID string