GH-4554 Correctly close changesets

- some shallow copies of Changeset were not closed at all

core/sail/base/src/main/java/org/eclipse/rdf4j/sail/base/Changeset.java

@@ -23,6 +23,7 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.concurrent.locks.StampedLock;
 import java.util.stream.Collectors;
@@ -121,6 +122,11 @@ public abstract class Changeset implements SailSink, ModelFactory {
 
 	private boolean closed;
 
+	/**
+	 * Reference counter for copies of this changeset.
+	 */
+	private AtomicInteger activeCopies = new AtomicInteger(1);
+
 	@Override
 	public void close() throws SailException {
 		closed = true;
@@ -131,23 +137,28 @@ public void close() throws SailException {
 		deprecatedContexts = null;
 		addedNamespaces = null;
 		removedPrefixes = null;
-		try {
-			if (approved instanceof AutoCloseable) {
-				((AutoCloseable) approved).close();
-			}
-		} catch (Exception e) {
-			throw new SailException(e);
-		} finally {
-			approved = null;
-			if (deprecated instanceof AutoCloseable) {
-				try {
-					((AutoCloseable) deprecated).close();
-				} catch (Exception e) {
-					throw new SailException(e);
-				} finally {
-					deprecated = null;
+		if (activeCopies.decrementAndGet() == 0) {
+			try {
+				if (approved instanceof AutoCloseable) {
+					((AutoCloseable) approved).close();
+				}
+			} catch (Exception e) {
+				throw new SailException(e);
+			} finally {
+				approved = null;
+				if (deprecated instanceof AutoCloseable) {
+					try {
+						((AutoCloseable) deprecated).close();
+					} catch (Exception e) {
+						throw new SailException(e);
+					} finally {
+						deprecated = null;
+					}
 				}
 			}
+		} else {
+			approved = null;
+			deprecated = null;
 		}
 	}
 
@@ -519,6 +530,9 @@ protected void setChangeset(Changeset from) {
 		assert !closed;
 		assert !from.closed;
 
+		this.activeCopies = from.activeCopies;
+		// count one more reference
+		this.activeCopies.incrementAndGet();
 		this.observed = from.observed;
 		this.approved = from.approved;
 		this.approvedEmpty = from.approvedEmpty;
@@ -550,7 +564,7 @@ public void flush() throws SailException {
 
 			@Override
 			public Model createEmptyModel() {
-				return Changeset.this.createEmptyModel();
+				throw new UnsupportedOperationException();
 			}
 		};
 

core/sail/base/src/main/java/org/eclipse/rdf4j/sail/base/SailSourceBranch.java

@@ -477,7 +477,10 @@ private void flush(SailSink sink) throws SailException {
 					&& !isChanged((Changeset) sink)) {
 				// one change to apply that is not in use to an empty Changeset
 				Changeset dst = (Changeset) sink;
-				dst.setChangeset(changes.pop());
+				Changeset src = changes.pop();
+				dst.setChangeset(src);
+				// correctly close changeset
+				src.close();
 			} else {
 				Iterator<Changeset> iter = changes.iterator();
 				while (iter.hasNext()) {
@@ -517,6 +520,9 @@ private void flush(Changeset change, SailSink sink) throws SailException {
 
 		change.sinkDeprecated(sink);
 		change.sinkApproved(sink);
+
+		// correctly close changeset
+		change.close();
 	}
 
 }