Merge pull request zendesk#192 from zendesk/ben/fix_memory_issues_again

Ben Osheroff · Ben Osheroff · commit a2e23a1ee1e7 · 2015-12-11T13:29:41.000-08:00
take another try at preventing memory issues.
diff --git a/src/main/java/com/zendesk/maxwell/MaxwellReplicator.java b/src/main/java/com/zendesk/maxwell/MaxwellReplicator.java
@@ -160,11 +160,11 @@ private RowMapBuffer getTransactionRows() throws Exception {
 
 		RowMapBuffer buffer = new RowMapBuffer(MAX_TX_ELEMENTS);
 
-		// currently to satisfy the test interface, the contract is to return null
-		// if the queue is empty.  should probably just replace this with an optional timeout...
-
 		while ( true ) {
 			v4Event = pollV4EventFromQueue();
+
+			// currently to satisfy the test interface, the contract is to return null
+			// if the queue is empty.  should probably just replace this with an optional timeout...
 			if (v4Event == null) {
 				ensureReplicatorThread();
 				continue;
diff --git a/src/main/java/com/zendesk/maxwell/util/ListWithDiskBuffer.java b/src/main/java/com/zendesk/maxwell/util/ListWithDiskBuffer.java
@@ -31,14 +31,14 @@ public void add(T element) throws IOException {
 			if ( file == null ) {
 				file = File.createTempFile("maxwell", "events");
 				file.deleteOnExit();
-				os = new ObjectOutputStream(new FileOutputStream(file));
-				is = new ObjectInputStream(new FileInputStream(file));
+				os = new ObjectOutputStream(new BufferedOutputStream(new FileOutputStream(file)));
 			}
 
 			if ( elementsInFile == 0 )
 				LOGGER.debug("Overflowed in-memory buffer, spilling over into " + file);
 
-			os.writeObject(this.list.removeFirst());
+			os.writeUnshared(this.list.removeFirst());
+
 			elementsInFile++;
 		}
 	}
@@ -52,8 +52,13 @@ public T getLast() {
 	}
 
 	public T removeFirst() throws IOException, ClassNotFoundException {
-		if ( elementsInFile > 0 && is != null ) {
-			T element = (T) is.readObject();
+		if ( elementsInFile > 0 ) {
+			if ( is == null ) {
+				os.flush();
+				is = new ObjectInputStream(new BufferedInputStream(new FileInputStream(file)));
+			}
+
+			T element = (T) is.readUnshared();
 			elementsInFile--;
 			return element;
 		} else {
