diff --git a/packages/integration-sdk-runtime/src/execution/__tests__/jobState.test.ts b/packages/integration-sdk-runtime/src/execution/__tests__/jobState.test.ts
index 267917cdf..b7874d26e 100644
--- a/packages/integration-sdk-runtime/src/execution/__tests__/jobState.test.ts
+++ b/packages/integration-sdk-runtime/src/execution/__tests__/jobState.test.ts
@@ -25,6 +25,7 @@ import {
 } from '../uploader';
 import { FlushedGraphObjectData } from '../../storage/types';
 import pMap from 'p-map';
+import { createMockIntegrationLogger } from '../../../test/util/fixtures';
 
 jest.mock('fs');
 
@@ -35,11 +36,14 @@ function entitiesToEntityKeySet(entities: Entity[]): Set<string> {
 }
 
 function createInMemoryStepGraphObjectDataUploaderCollector(
-  partial?: CreateQueuedStepGraphObjectDataUploaderParams,
+  partial?: Partial<CreateQueuedStepGraphObjectDataUploaderParams>,
 ) {
   const graphObjectDataCollection: FlushedGraphObjectData[] = [];
 
+  const logger = createMockIntegrationLogger();
+
   const uploader = createQueuedStepGraphObjectDataUploader({
+    logger,
     stepId: uuid(),
     uploadConcurrency: 5,
     upload(graphObjectData) {
@@ -307,8 +311,10 @@ describe('upload callbacks', () => {
 
   test('#waitUntilUploadsComplete should resolve when all uploads completed', async () => {
     const graphObjectDataCollection: FlushedGraphObjectData[] = [];
+    const logger = createMockIntegrationLogger();
 
     const uploader = createQueuedStepGraphObjectDataUploader({
+      logger,
       stepId: uuid(),
       uploadConcurrency: 5,
       async upload(graphObjectData) {
diff --git a/packages/integration-sdk-runtime/src/execution/executeIntegration.ts b/packages/integration-sdk-runtime/src/execution/executeIntegration.ts
index c26837c4c..9600e06be 100644
--- a/packages/integration-sdk-runtime/src/execution/executeIntegration.ts
+++ b/packages/integration-sdk-runtime/src/execution/executeIntegration.ts
@@ -250,7 +250,9 @@ export async function executeWithContext<
       }
 
       const {
-        graphObjectStore = new FileSystemGraphObjectStore(),
+        graphObjectStore = new FileSystemGraphObjectStore({
+          logger,
+        }),
         createStepGraphObjectDataUploader,
         resultsCallback,
       } = options;
diff --git a/packages/integration-sdk-runtime/src/execution/uploader.test.ts b/packages/integration-sdk-runtime/src/execution/uploader.test.ts
index d7252d5b9..97c097c6e 100644
--- a/packages/integration-sdk-runtime/src/execution/uploader.test.ts
+++ b/packages/integration-sdk-runtime/src/execution/uploader.test.ts
@@ -52,7 +52,10 @@ describe('#createQueuedStepGraphObjectDataUploader', () => {
     const uploaded: FlushedGraphObjectData[] = [];
     let numQueued = 0;
 
+    const logger = createMockIntegrationLogger();
+
     const uploader = createQueuedStepGraphObjectDataUploader({
+      logger,
       stepId: uuid(),
       uploadConcurrency: Infinity,
       async upload(d) {
@@ -77,7 +80,10 @@ describe('#createQueuedStepGraphObjectDataUploader', () => {
     const uploaded: FlushedGraphObjectData[] = [];
     let throttleCount = 0;
 
+    const logger = createMockIntegrationLogger();
+
     const uploader = createQueuedStepGraphObjectDataUploader({
+      logger,
       stepId: uuid(),
       uploadConcurrency: 2,
       async upload(d) {
@@ -101,7 +107,10 @@ describe('#createQueuedStepGraphObjectDataUploader', () => {
 
     let numQueued = 0;
 
+    const logger = createMockIntegrationLogger();
+
     const uploader = createQueuedStepGraphObjectDataUploader({
+      logger,
       stepId,
       uploadConcurrency: 2,
       async upload(d) {
@@ -143,7 +152,10 @@ describe('#createQueuedStepGraphObjectDataUploader', () => {
 
     let numQueued = 0;
 
+    const logger = createMockIntegrationLogger();
+
     const uploader = createQueuedStepGraphObjectDataUploader({
+      logger,
       stepId,
       uploadConcurrency: 2,
       async upload(d) {
diff --git a/packages/integration-sdk-runtime/src/execution/uploader.ts b/packages/integration-sdk-runtime/src/execution/uploader.ts
index ea0505b0c..2c33955b4 100644
--- a/packages/integration-sdk-runtime/src/execution/uploader.ts
+++ b/packages/integration-sdk-runtime/src/execution/uploader.ts
@@ -1,4 +1,7 @@
-import { UploadError } from '@jupiterone/integration-sdk-core';
+import {
+  IntegrationLogger,
+  UploadError,
+} from '@jupiterone/integration-sdk-core';
 import PQueue from 'p-queue/dist';
 import { FlushedGraphObjectData } from '../storage/types';
 import {
@@ -19,6 +22,7 @@ export type CreateStepGraphObjectDataUploaderFunction = (
 ) => StepGraphObjectDataUploader;
 
 export interface CreateQueuedStepGraphObjectDataUploaderParams {
+  logger: IntegrationLogger;
   stepId: string;
   uploadConcurrency: number;
   upload: (graphObjectData: FlushedGraphObjectData) => Promise<void>;
@@ -26,6 +30,7 @@ export interface CreateQueuedStepGraphObjectDataUploaderParams {
 }
 
 export function createQueuedStepGraphObjectDataUploader({
+  logger,
   stepId,
   uploadConcurrency: maximumQueueSize,
   upload,
@@ -69,6 +74,10 @@ export function createQueuedStepGraphObjectDataUploader({
       queue
         .add(() => upload(graphObjectData))
         .catch((err) => {
+          logger.warn(
+            { err, stepId, graphObjectData },
+            'Error uploading graph object data batch',
+          );
           // Do not pause the queue entirely. We will try to prevent additional
           // tasks from being added to the queue, but even if an error occurs,
           // we should try uploading the remaining data that we have queued up.
@@ -100,6 +109,15 @@ export function createQueuedStepGraphObjectDataUploader({
         // this time, we could be receiving additional tasks in our queue that
         // will grow the queue.
         completed = true;
+
+        logger.debug(
+          {
+            stepId,
+            uploadErrorCount: uploadErrors.length,
+            typesInvolvedInFailures: Array.from(typesInvolvedInFailures),
+          },
+          'Upload queue processing complete',
+        );
       }
 
       if (uploadErrors.length) {
@@ -150,6 +168,7 @@ export function createPersisterApiStepGraphObjectDataUploader({
   uploadBatchSizeInBytes = DEFAULT_UPLOAD_BATCH_SIZE_IN_BYTES,
 }: CreatePersisterApiStepGraphObjectDataUploaderParams) {
   return createQueuedStepGraphObjectDataUploader({
+    logger: synchronizationJobContext.logger,
     stepId,
     uploadConcurrency,
     async upload(graphObjectData) {
diff --git a/packages/integration-sdk-runtime/src/storage/FileSystemGraphObjectStore/FileSystemGraphObjectStore.ts b/packages/integration-sdk-runtime/src/storage/FileSystemGraphObjectStore/FileSystemGraphObjectStore.ts
index 28f3f3471..575183519 100644
--- a/packages/integration-sdk-runtime/src/storage/FileSystemGraphObjectStore/FileSystemGraphObjectStore.ts
+++ b/packages/integration-sdk-runtime/src/storage/FileSystemGraphObjectStore/FileSystemGraphObjectStore.ts
@@ -10,6 +10,7 @@ import {
   GetIndexMetadataForGraphObjectTypeParams,
   IntegrationStep,
   GraphObjectIterateeOptions,
+  IntegrationLogger,
 } from '@jupiterone/integration-sdk-core';
 
 import { flushDataToDisk } from './flushDataToDisk';
@@ -52,6 +53,11 @@ export interface FileSystemGraphObjectStoreParams {
    * Whether the files that are written to disk should be minified or not
    */
   prettifyFiles?: boolean;
+
+  /**
+   * Optional logger for debugging and tracking data flow
+   */
+  logger?: IntegrationLogger;
 }
 
 interface GraphObjectIndexMetadataMap {
@@ -135,6 +141,7 @@ export class FileSystemGraphObjectStore implements GraphObjectStore {
     string,
     GraphObjectLocationOnDisk
   >(ENTITY_LOCATION_ON_DISK_DEFAULT_MAP_KEY_SPACE);
+  private readonly logger?: IntegrationLogger;
 
   constructor(params?: FileSystemGraphObjectStoreParams) {
     this.semaphore = new Sema(BINARY_SEMAPHORE_CONCURRENCY);
@@ -142,6 +149,7 @@ export class FileSystemGraphObjectStore implements GraphObjectStore {
       params?.graphObjectFileSize || DEFAULT_GRAPH_OBJECT_FILE_SIZE;
 
     this.prettifyFiles = params?.prettifyFiles || false;
+    this.logger = params?.logger;
     this.graphObjectBufferThresholdInBytes = min([
       params?.graphObjectBufferThresholdInBytes ||
         DEFAULT_UPLOAD_BATCH_SIZE_IN_BYTES,
@@ -325,28 +333,64 @@ export class FileSystemGraphObjectStore implements GraphObjectStore {
         });
 
         if (indexable.length) {
-          await Promise.all(
-            chunk(indexable, this.graphObjectFileSize).map(async (data) => {
-              const graphObjectsToFilePaths = await flushDataToDisk({
-                storageDirectoryPath: stepId,
-                collectionType: 'entities',
-                data,
-                pretty: this.prettifyFiles,
-              });
-
-              for (const {
-                graphDataPath,
-                collection,
-              } of graphObjectsToFilePaths) {
-                for (const [index, e] of collection.entries()) {
-                  this.entityOnDiskLocationMap.set(e._key, {
-                    graphDataPath,
-                    index,
-                  });
-                }
-              }
-            }),
+          const chunks = chunk(indexable, this.graphObjectFileSize);
+          this.logger?.debug(
+            {
+              stepId,
+              entityCount: indexable.length,
+              chunkCount: chunks.length,
+              chunkSize: this.graphObjectFileSize,
+            },
+            'Flushing entity chunks to disk',
           );
+
+          try {
+            await Promise.all(
+              chunks.map(async (data, chunkIndex) => {
+                const graphObjectsToFilePaths = await flushDataToDisk({
+                  storageDirectoryPath: stepId,
+                  collectionType: 'entities',
+                  data,
+                  pretty: this.prettifyFiles,
+                  logger: this.logger,
+                });
+
+                for (const {
+                  graphDataPath,
+                  collection,
+                } of graphObjectsToFilePaths) {
+                  for (const [index, e] of collection.entries()) {
+                    this.entityOnDiskLocationMap.set(e._key, {
+                      graphDataPath,
+                      index,
+                    });
+                  }
+                }
+
+                this.logger?.debug(
+                  {
+                    stepId,
+                    chunkIndex,
+                    entitiesInChunk: data.length,
+                    filesCreated: graphObjectsToFilePaths.length,
+                  },
+                  'Entity chunk flushed successfully',
+                );
+              }),
+            );
+          } catch (error) {
+            this.logger?.error(
+              {
+                stepId,
+                entityCount: indexable.length,
+                chunkCount: chunks.length,
+                error: error.message,
+                errorStack: error.stack,
+              },
+              'Failed to flush entity chunks to disk',
+            );
+            throw error;
+          }
         }
 
         this.localGraphObjectStore.flushEntities(entities, stepId);
@@ -354,7 +398,18 @@ export class FileSystemGraphObjectStore implements GraphObjectStore {
       }
 
       if (onEntitiesFlushed) {
-        await onEntitiesFlushed(entitiesToUpload);
+        try {
+          await onEntitiesFlushed(entitiesToUpload);
+        } catch (err) {
+          this.logger?.error(
+            {
+              entityCount: entitiesToUpload.length,
+              err,
+            },
+            'onEntitiesFlushed callback failed',
+          );
+          throw err;
+        }
       }
     });
   }
@@ -406,16 +461,51 @@ export class FileSystemGraphObjectStore implements GraphObjectStore {
         });
 
         if (indexable.length) {
-          await Promise.all(
-            chunk(indexable, this.graphObjectFileSize).map(async (data) => {
-              await flushDataToDisk({
-                storageDirectoryPath: stepId,
-                collectionType: 'relationships',
-                data,
-                pretty: this.prettifyFiles,
-              });
-            }),
+          const chunks = chunk(indexable, this.graphObjectFileSize);
+          this.logger?.debug(
+            {
+              stepId,
+              relationshipCount: indexable.length,
+              chunkCount: chunks.length,
+              chunkSize: this.graphObjectFileSize,
+            },
+            'Flushing relationship chunks to disk',
           );
+
+          try {
+            await Promise.all(
+              chunks.map(async (data, chunkIndex) => {
+                await flushDataToDisk({
+                  storageDirectoryPath: stepId,
+                  collectionType: 'relationships',
+                  data,
+                  pretty: this.prettifyFiles,
+                  logger: this.logger,
+                });
+
+                this.logger?.debug(
+                  {
+                    stepId,
+                    chunkIndex,
+                    relationshipsInChunk: data.length,
+                  },
+                  'Relationship chunk flushed successfully',
+                );
+              }),
+            );
+          } catch (error) {
+            this.logger?.error(
+              {
+                stepId,
+                relationshipCount: indexable.length,
+                chunkCount: chunks.length,
+                error: error.message,
+                errorStack: error.stack,
+              },
+              'Failed to flush relationship chunks to disk',
+            );
+            throw error;
+          }
         }
 
         this.localGraphObjectStore.flushRelationships(relationships, stepId);
@@ -423,7 +513,19 @@ export class FileSystemGraphObjectStore implements GraphObjectStore {
       }
 
       if (onRelationshipsFlushed) {
-        await onRelationshipsFlushed(relationshipsToUpload);
+        try {
+          await onRelationshipsFlushed(relationshipsToUpload);
+        } catch (error) {
+          this.logger?.error(
+            {
+              relationshipCount: relationshipsToUpload.length,
+              error: error.message,
+              errorStack: error.stack,
+            },
+            'onRelationshipsFlushed callback failed',
+          );
+          throw error;
+        }
       }
     });
   }
diff --git a/packages/integration-sdk-runtime/src/storage/FileSystemGraphObjectStore/flushDataToDisk.ts b/packages/integration-sdk-runtime/src/storage/FileSystemGraphObjectStore/flushDataToDisk.ts
index f83a86d7a..061ce80ea 100644
--- a/packages/integration-sdk-runtime/src/storage/FileSystemGraphObjectStore/flushDataToDisk.ts
+++ b/packages/integration-sdk-runtime/src/storage/FileSystemGraphObjectStore/flushDataToDisk.ts
@@ -17,6 +17,7 @@ interface FlushDataToDiskInput<TGraphObject = Entity | Relationship> {
   collectionType: CollectionType;
   data: TGraphObject[];
   pretty?: boolean;
+  logger?: any;
 }
 
 interface GraphObjectToFilePath<TGraphObject = Entity | Relationship> {
@@ -34,15 +35,33 @@ export async function flushDataToDisk<TGraphObject = Entity | Relationship>({
   collectionType,
   data,
   pretty,
+  logger,
 }: FlushDataToDiskInput<TGraphObject>): Promise<
   GraphObjectToFilePath<TGraphObject>[]
 > {
   // split the data by type first
   const groupedCollections = groupBy(data, '_type');
 
+  const totalObjects = data.length;
+  const typeCount = Object.keys(groupedCollections).length;
+  const chunkInfo = Object.entries(groupedCollections)
+    .map(([type, items]) => `${type}:${items.length}`)
+    .join(', ');
+
+  logger?.debug(
+    {
+      stepId: storageDirectoryPath,
+      collectionType,
+      totalObjects,
+      typeCount,
+      chunkInfo,
+    },
+    'Flushing data to disk',
+  );
+
   // for each collection, write the data to disk,
   // then symlink to index directory
-  return await pMap(
+  const results = await pMap(
     Object.entries(groupedCollections),
     async ([type, collection]) => {
       const filename = generateJsonFilename();
@@ -53,25 +72,65 @@ export async function flushDataToDisk<TGraphObject = Entity | Relationship>({
       });
       const indexPath = buildIndexFilePath({ type, collectionType, filename });
 
-      await writeJsonToPath({
-        path: graphDataPath,
-        data: {
-          [collectionType]: collection,
-        },
-        pretty,
-      });
+      try {
+        await writeJsonToPath({
+          path: graphDataPath,
+          data: {
+            [collectionType]: collection,
+          },
+          pretty,
+        });
 
-      await symlink({
-        sourcePath: graphDataPath,
-        destinationPath: indexPath,
-      });
-      return {
-        graphDataPath,
-        collection,
-      };
+        await symlink({
+          sourcePath: graphDataPath,
+          destinationPath: indexPath,
+        });
+
+        logger?.debug(
+          {
+            type,
+            count: collection.length,
+            filename,
+            graphDataPath,
+          },
+          'Chunk written to disk successfully',
+        );
+
+        return {
+          graphDataPath,
+          collection,
+        };
+      } catch (error) {
+        logger?.error(
+          {
+            error,
+            type,
+            count: collection.length,
+            filename,
+            graphDataPath,
+          },
+          'Failed to write chunk to disk',
+        );
+        throw error;
+      }
     },
     { concurrency: 3 },
   );
+
+  logger?.info(
+    {
+      stepId: storageDirectoryPath,
+      collectionType,
+      chunksWritten: results.length,
+      totalObjectsWritten: results.reduce(
+        (sum, r) => sum + r.collection.length,
+        0,
+      ),
+    },
+    'Successfully flushed all chunks to disk',
+  );
+
+  return results;
 }
 
 function generateJsonFilename() {