broker/client/retry_reader.go

@@ -91,7 +91,7 @@ func (rr *RetryReader) Read(p []byte) (n int, err error) {
 		var squelch bool
 
 		switch err {
-		case context.DeadlineExceeded, context.Canceled, ErrJournalNotFound:
+		case context.DeadlineExceeded, context.Canceled:
 			return // Surface to caller.
 		case ErrOffsetNotYetAvailable:
 			if rr.Reader.Request.Block {
@@ -104,6 +104,14 @@ func (rr *RetryReader) Read(p []byte) (n int, err error) {
 			// Suppress logging for expected errors on first read attempt.
 			// We may be racing a concurrent Etcd watch and assignment of the broker cluster.
 			squelch = attempt == 0
+		case ErrJournalNotFound:
+			// If a Header was attached to the request, the journal was not found
+			// even after honoring its read-through Etcd revision. If not,
+			// we allow a handful of attempts to work around expected propagation
+			// delays of Etcd revisions if the journal was just created.
+			if rr.Reader.Request.Header != nil || attempt > 3 {
+				return // Surface to caller.
+			}
 		case io.EOF:
 			// EOF means we had an established RPC, but it might not have sent
 			// any data before being closed server-side, so clear `attempts` to