@@ -64,26 +64,42 @@ private double ingestSingleSource(String fileId, MLBatchIngestionInput mlBatchIn
6464 connection .setRequestMethod ("GET" );
6565 connection .setRequestProperty ("Authorization" , "Bearer " + apiKey );
6666
67- InputStreamReader inputStreamReader = AccessController
68- .doPrivileged ((PrivilegedExceptionAction <InputStreamReader >) () -> new InputStreamReader (connection .getInputStream ()));
69- BufferedReader reader = new BufferedReader (inputStreamReader );
70-
71- List <String > linesBuffer = new ArrayList <>();
72- String line ;
73- int lineCount = 0 ;
74- // Atomic counters for tracking success and failure
75- AtomicInteger successfulBatches = new AtomicInteger (0 );
76- AtomicInteger failedBatches = new AtomicInteger (0 );
77- // List of CompletableFutures to track batch ingestion operations
78- List <CompletableFuture <Void >> futures = new ArrayList <>();
79-
80- while ((line = reader .readLine ()) != null ) {
81- linesBuffer .add (line );
82- lineCount ++;
83-
84- // Process every 100 lines
85- if (lineCount == 100 ) {
86- // Create a CompletableFuture that will be completed by the bulkResponseListener
67+ try (
68+ InputStreamReader inputStreamReader = AccessController
69+ .doPrivileged ((PrivilegedExceptionAction <InputStreamReader >) () -> new InputStreamReader (connection .getInputStream ()));
70+ BufferedReader reader = new BufferedReader (inputStreamReader )
71+ ) {
72+ List <String > linesBuffer = new ArrayList <>();
73+ String line ;
74+ int lineCount = 0 ;
75+ // Atomic counters for tracking success and failure
76+ AtomicInteger successfulBatches = new AtomicInteger (0 );
77+ AtomicInteger failedBatches = new AtomicInteger (0 );
78+ // List of CompletableFutures to track batch ingestion operations
79+ List <CompletableFuture <Void >> futures = new ArrayList <>();
80+
81+ while ((line = reader .readLine ()) != null ) {
82+ linesBuffer .add (line );
83+ lineCount ++;
84+
85+ // Process every 100 lines
86+ if (lineCount % 100 == 0 ) {
87+ // Create a CompletableFuture that will be completed by the bulkResponseListener
88+ CompletableFuture <Void > future = new CompletableFuture <>();
89+ batchIngest (
90+ linesBuffer ,
91+ mlBatchIngestionInput ,
92+ getBulkResponseListener (successfulBatches , failedBatches , future ),
93+ sourceIndex ,
94+ isSoleSource
95+ );
96+
97+ futures .add (future );
98+ linesBuffer .clear ();
99+ }
100+ }
101+ // Process any remaining lines in the buffer
102+ if (!linesBuffer .isEmpty ()) {
87103 CompletableFuture <Void > future = new CompletableFuture <>();
88104 batchIngest (
89105 linesBuffer ,
@@ -92,32 +108,17 @@ private double ingestSingleSource(String fileId, MLBatchIngestionInput mlBatchIn
92108 sourceIndex ,
93109 isSoleSource
94110 );
95-
96111 futures .add (future );
97- linesBuffer .clear ();
98- lineCount = 0 ;
99112 }
100- }
101- // Process any remaining lines in the buffer
102- if (!linesBuffer .isEmpty ()) {
103- CompletableFuture <Void > future = new CompletableFuture <>();
104- batchIngest (
105- linesBuffer ,
106- mlBatchIngestionInput ,
107- getBulkResponseListener (successfulBatches , failedBatches , future ),
108- sourceIndex ,
109- isSoleSource
110- );
111- futures .add (future );
112- }
113113
114- reader .close ();
115- // Combine all futures and wait for completion
116- CompletableFuture <Void > allFutures = CompletableFuture .allOf (futures .toArray (new CompletableFuture [0 ]));
117- // Wait for all tasks to complete
118- allFutures .join ();
119- int totalBatches = successfulBatches .get () + failedBatches .get ();
120- successRate = (double ) successfulBatches .get () / totalBatches * 100 ;
114+ reader .close ();
115+ // Combine all futures and wait for completion
116+ CompletableFuture <Void > allFutures = CompletableFuture .allOf (futures .toArray (new CompletableFuture [0 ]));
117+ // Wait for all tasks to complete
118+ allFutures .join ();
119+ int totalBatches = successfulBatches .get () + failedBatches .get ();
120+ successRate = (totalBatches == 0 ) ? 100 : (double ) successfulBatches .get () / totalBatches * 100 ;
121+ }
121122 } catch (PrivilegedActionException e ) {
122123 throw new RuntimeException ("Failed to read from OpenAI file API: " , e );
123124 } catch (Exception e ) {
0 commit comments