[JENKINS-67587] Create build scoped connection broken flag

.gitignore

@@ -13,3 +13,5 @@ work
 .classpath
 .project
 .settings/
+
+.DS_Store

pom.xml

@@ -218,6 +218,24 @@
   <build>
     <defaultGoal>clean verify</defaultGoal>
     <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <version>3.0.0-M3</version>
+        <configuration>
+          <useSystemClassLoader>false</useSystemClassLoader>
+        </configuration>
+      </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-site-plugin</artifactId>
+        <version>3.7.1</version>
+      </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-surefire-report-plugin</artifactId>
+        <version>3.0.0-M5</version>
+      </plugin>
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-failsafe-plugin</artifactId>

src/main/java/jenkins/plugins/logstash/LogstashOutputStream.java

@@ -29,6 +29,9 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
 /**
  * Output stream that writes each line to the provided delegate output stream
@@ -38,13 +41,28 @@
  * @author Rusty Gerard
  */
 public class LogstashOutputStream extends LineTransformationOutputStream {
+  private static final Logger LOGGER = Logger.getLogger(LogstashOutputStream.class.getName());
+
   private final OutputStream delegate;
   private final LogstashWriter logstash;
+  private final AtomicBoolean isBuildConnectionBroken;
+  private final String run;
 
   public LogstashOutputStream(OutputStream delegate, LogstashWriter logstash) {
+    this(delegate, logstash, new AtomicBoolean(false), "");
+  }
+
+  public LogstashOutputStream(OutputStream delegate, LogstashWriter logstash, AtomicBoolean isBuildConnectionBroken, String run) {
     super();
     this.delegate = delegate;
     this.logstash = logstash;
+    this.isBuildConnectionBroken = isBuildConnectionBroken;
+    this.run = run;
+
+  }
+
+  public AtomicBoolean getIsBuildConnectionBroken() {
+    return isBuildConnectionBroken;
   }
 
   // for testing purposes
@@ -58,11 +76,19 @@ protected void eol(byte[] b, int len) throws IOException {
     delegate.write(b, 0, len);
     this.flush();
 
-    if(!logstash.isConnectionBroken()) {
-      String line = new String(b, 0, len, logstash.getCharset());
-      line = ConsoleNote.removeNotes(line).trim();
-      logstash.write(line);
+    if (!getIsBuildConnectionBroken().get()) {
+      if (!logstash.isConnectionBroken()) {
+        String line = new String(b, 0, len, logstash.getCharset());
+        line = ConsoleNote.removeNotes(line).trim();
+        logstash.write(line);
+      }
+      // Once it gets connection broken, set the build connection flag to true.
+      if (logstash.isConnectionBroken()) {
+        getIsBuildConnectionBroken().set(true);
+        LOGGER.log(Level.WARNING, "Mark logstash connection broken for build: {0}.", run);
+      }
     }
+
   }
 
   /**

src/main/java/jenkins/plugins/logstash/pipeline/GlobalDecorator.java

@@ -3,6 +3,7 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -25,6 +26,7 @@ public class GlobalDecorator extends TaskListenerDecorator {
   private transient Run<?, ?> run;
   private String stageName;
   private String agentName;
+  AtomicBoolean isBuildScopedConnectionBroken;
 
   public GlobalDecorator(WorkflowRun run) {
     this(run, null, null);
@@ -34,12 +36,13 @@ public GlobalDecorator(WorkflowRun run, String stageName, String agentName) {
     this.run = run;
     this.stageName = stageName;
     this.agentName = agentName;
+    this.isBuildScopedConnectionBroken = new AtomicBoolean(false);
   }
 
   @Override
   public OutputStream decorate(OutputStream logger) throws IOException, InterruptedException {
     LogstashWriter writer = new LogstashWriter(run, logger, null, StandardCharsets.UTF_8, stageName, agentName);
-    LogstashOutputStream out = new LogstashOutputStream(logger, writer);
+    LogstashOutputStream out = new LogstashOutputStream(logger, writer, isBuildScopedConnectionBroken, run.toString());
     return out;
   }
 

src/test/java/jenkins/plugins/logstash/LogstashOutputStreamTest.java

@@ -8,6 +8,7 @@
 import java.io.ByteArrayOutputStream;
 import java.io.OutputStream;
 import java.nio.charset.Charset;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.junit.After;
 import org.junit.Before;
@@ -61,7 +62,7 @@ public void eolSuccess() throws Exception {
 
     // Verify results
     assertEquals("Results don't match", msg, buffer.toString());
-    verify(mockWriter).isConnectionBroken();
+    verify(mockWriter, times(2)).isConnectionBroken();
     verify(mockWriter).write(msg);
     verify(mockWriter).getCharset();
   }
@@ -104,14 +105,14 @@ public void eolSuccessConnectionBroken() throws Exception {
 
     //Verify calls were made to the dao logging twice, not three times.
     verify(mockWriter, times(2)).write(msg);
-    verify(mockWriter, times(3)).isConnectionBroken();
+    verify(mockWriter, times(6)).isConnectionBroken();
     verify(mockWriter, times(2)).getCharset();
   }
 
   @Test
   public void eolSuccessNoDao() throws Exception {
     when(mockWriter.isConnectionBroken()).thenReturn(true);
-    LogstashOutputStream los = new LogstashOutputStream(buffer, mockWriter);
+    LogstashOutputStream los = new LogstashOutputStream(buffer, mockWriter, new AtomicBoolean(false),"");
     String msg = "test";
     buffer.reset();
 
@@ -120,6 +121,6 @@ public void eolSuccessNoDao() throws Exception {
 
     // Verify results
     assertEquals("Results don't match", msg, buffer.toString());
-    verify(mockWriter).isConnectionBroken();
+    verify(mockWriter, times(2)).isConnectionBroken();
   }
 }