Check stream message is confirmed

Author: acogoluegnes

Diff for: go-stream/offset_tracking_receive.go

@@ -40,7 +40,7 @@ func main() {
 		if string(message.GetData()) == "marker" {
 			lastOffset.Store(consumerContext.Consumer.GetOffset())
 			_ = consumerContext.Consumer.StoreOffset()
-            _ = consumerContext.Consumer.Close()
+			_ = consumerContext.Consumer.Close()
 			ch <- true
 		}
 	}

Diff for: go-stream/offset_tracking_send.go

@@ -54,10 +54,12 @@ func handlePublishConfirm(confirms stream.ChannelPublishConfirm, messageCount in
 	go func() {
 		confirmedCount := 0
 		for confirmed := range confirms {
-			for _, _ = range confirmed {
-				confirmedCount++
-				if confirmedCount == messageCount {
-					ch <- true
+			for _, msg := range confirmed {
+				if msg.IsConfirmed() {
+					confirmedCount++
+					if confirmedCount == messageCount {
+						ch <- true
+					}
 				}
 			}
 		}

Diff for: java-stream-mvn/src/main/java/OffsetTrackingSend.java

@@ -23,7 +23,11 @@ public static void main(String[] args) throws InterruptedException {
       IntStream.range(0, messageCount).forEach(i -> {
         String body = i == messageCount - 1 ? "marker" : "hello";
         producer.send(producer.messageBuilder().addData(body.getBytes(UTF_8)).build(),
-            ctx -> confirmedLatch.countDown());
+            ctx -> {
+              if (ctx.isConfirmed()) {
+                confirmedLatch.countDown();
+              }
+            });
       });
 
       boolean completed = confirmedLatch.await(60, TimeUnit.SECONDS);