[proxima-beam-core] refactor inflight writes locking

Author: je-ik

beam/core/src/main/java/cz/o2/proxima/beam/io/ProximaIO.java

@@ -177,12 +177,12 @@ public void processElement(@Element StreamElement element) {
       boolean isTransactional =
           element.getAttributeDescriptor().getTransactionMode() != TransactionMode.NONE;
       int weight = isTransactional ? transactionalWriteWeight : 1;
-      synchronized (pendingWrites) {
-        while (inflightWriteWeights.get() >= maxPendingWrites) {
+      while (inflightWriteWeights.get() + weight >= maxPendingWrites) {
+        synchronized (pendingWrites) {
           ExceptionUtils.unchecked(() -> pendingWrites.wait(100));
         }
-        inflightWriteWeights.addAndGet(weight);
       }
+      inflightWriteWeights.addAndGet(weight);
       writeRunnableRef.set(
           () -> {
             CompletableFuture<Pair<Boolean, Throwable>> writeResult = new CompletableFuture<>();