NakadiReader implements IORunnable { void runInternal() throws IOException, BackoffException { LOG.info("Starting to listen for events for {}", eventNames); JsonInput jsonInput = openJsonInput(); int errorCount = 0; while (true) { try { final JsonParser jsonParser = jsonInput.getJsonParser(); if (Thread.currentThread().isInterrupted()) { throw new InterruptedIOException("Interrupted"); } readBatch(jsonParser); errorCount = 0; } catch (IOException e) { final boolean wasInterrupted = Thread.currentThread().isInterrupted(); metricsCollector.markErrorWhileConsuming(); if (errorCount > 0) { LOG.warn("Got [{}] [{}] while reading events for {} after [{}] retries", e.getClass().getSimpleName(), e.getMessage(), eventNames, errorCount, e); } else { LOG.info("Got [{}] [{}] while reading events for {}", e.getClass().getSimpleName(), e.getMessage(), eventNames, e); } jsonInput.close(); if (wasInterrupted || Thread.currentThread().isInterrupted()) { LOG.warn("Thread was interrupted"); break; } try { LOG.debug("Reconnecting after [{}] errors", errorCount); jsonInput = backoffStrategy.call(errorCount, e, this::openJsonInput); LOG.info("Reconnected after [{}] errors", errorCount); metricsCollector.markReconnection(); } catch (InterruptedException interruptedException) { LOG.warn("Interrupted during reconnection", interruptedException); Thread.currentThread().interrupt(); return; } errorCount++; } catch (Throwable e) { LOG.warn("Got [{}] [{}] while reading events for {}", e.getClass().getSimpleName(), e.getMessage(), eventNames, e); try { jsonInput.close(); } catch (Throwable suppressed) { e.addSuppressed(e); } throw e; } } } NakadiReader(URI uri, RequestFactory requestFactory, BackoffStrategy backoffStrategy, CursorManager cursorManager, ObjectMapper objectMapper, Set<String> eventNames, Optional<Subscription> subscription, Optional<Lock> lock, Class<T> eventClass, Listener<T> listener);   NakadiReader(URI uri, RequestFactory requestFactory, BackoffStrategy backoffStrategy, CursorManager cursorManager, Set<String> eventNames, Optional<Subscription> subscription, Optional<Lock> lock, EventReader<T> eventReader, Listener<T> listener, BatchHandler batchHandler, final MetricsCollector metricsCollector); @Override void run();  }
@Test public void shouldNotReconnectWithoutBackoff() throws IOException, InterruptedException, BackoffException { final Response response = mock(Response.class); final ByteArrayInputStream emptyInputStream = new ByteArrayInputStream(new byte[0]); when(response.getBody()).thenReturn(emptyInputStream); final Request request = mock(Request.class); when(request.execute()).thenReturn(response); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final NoBackoffStrategy backoffStrategy = new NoBackoffStrategy(); final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener); expectedException.expect(BackoffException.class); expectedException.expect(ComposeMatchers.hasFeature("retries", BackoffException::getRetries, equalTo(0))); expectedException.expectCause(ComposeMatchers.hasFeature("message", Exception::getMessage, Matchers.containsString("Stream was closed"))); nakadiReader.runInternal(); }
@Test public void shouldHandleBrokenInput() throws IOException, InterruptedException, BackoffException { final Response response = mock(Response.class); final ByteArrayInputStream initialInputStream = new ByteArrayInputStream("{\"".getBytes("utf-8")); when(response.getBody()).thenReturn(initialInputStream); final Request request = mock(Request.class); when(request.execute()).thenReturn(response); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final NoBackoffStrategy backoffStrategy = new NoBackoffStrategy(); final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener); expectedException.expect(BackoffException.class); expectedException.expect(ComposeMatchers.hasFeature(BackoffException::getRetries, equalTo(0))); expectedException.expectCause(instanceOf(JsonProcessingException.class)); expectedException.expectCause(ComposeMatchers.hasFeature("message", Exception::getMessage, Matchers.containsString("Unexpected end-of-input"))); nakadiReader.runInternal(); }
@Test public void shouldHandleBrokenInputInEvents() throws IOException, InterruptedException, BackoffException { final Response response = mock(Response.class); final ByteArrayInputStream initialInputStream = new ByteArrayInputStream("{\"cursor\":{\"partition\":\"123\",\"offset\":\"456\"},\"events\":[{\"id\":".getBytes("utf-8")); when(response.getBody()).thenReturn(initialInputStream); final Request request = mock(Request.class); when(request.execute()).thenReturn(response); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final NoBackoffStrategy backoffStrategy = new NoBackoffStrategy(); final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener); expectedException.expect(BackoffException.class); expectedException.expect(ComposeMatchers.hasFeature(BackoffException::getRetries, equalTo(0))); expectedException.expectCause(instanceOf(JsonProcessingException.class)); expectedException.expectCause(ComposeMatchers.hasFeature("message", Exception::getMessage, Matchers.containsString("Unexpected end-of-input"))); nakadiReader.runInternal(); }
@Test public void shouldRetryConnectionOnEof() throws IOException, InterruptedException, BackoffException { final Response response = mock(Response.class); final ByteArrayInputStream initialInputStream = new ByteArrayInputStream("{\"cursor\":{\"partition\":\"0\",\"offset\":\"0\"}}".getBytes("utf-8")); when(response.getBody()).thenReturn(initialInputStream); final Request request = mock(Request.class); when(request.execute()).thenReturn(response).thenThrow(new IOException("Reconnection failed")); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final ExponentialBackoffStrategy backoffStrategy = new EqualJitterBackoffStrategy(1, 1, 2, 1); final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener); expectedException.expect(BackoffException.class); expectedException.expect(ComposeMatchers.hasFeature(BackoffException::getRetries, equalTo(1))); expectedException.expectCause(instanceOf(IOException.class)); expectedException.expectCause(ComposeMatchers.hasFeature("message", Exception::getMessage, Matchers.containsString("Reconnection failed"))); nakadiReader.runInternal(); }
@Test public void shouldRetryConnectionMultipleTimesOnEof() throws IOException, InterruptedException, BackoffException { final Response response = mock(Response.class); final ByteArrayInputStream initialInputStream = new ByteArrayInputStream("{\"cursor\":{\"partition\":\"0\",\"offset\":\"0\"}}".getBytes("utf-8")); when(response.getBody()).thenReturn(initialInputStream); final Request request = mock(Request.class); when(request.execute()).thenReturn(response).thenThrow(new IOException("Reconnection failed")); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final ExponentialBackoffStrategy backoffStrategy = new EqualJitterBackoffStrategy(1, 1, 2, 4); final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener); expectedException.expect(BackoffException.class); expectedException.expect(ComposeMatchers.hasFeature(BackoffException::getRetries, equalTo(4))); expectedException.expectCause(instanceOf(IOException.class)); expectedException.expectCause(ComposeMatchers.hasFeature("message", Exception::getMessage, Matchers.containsString("Reconnection failed"))); nakadiReader.runInternal(); }
@Test public void shouldRetryConnectionAfterExceptionDuringReconnection() throws IOException, InterruptedException, BackoffException { final Response response = mock(Response.class); final ByteArrayInputStream initialInputStream = new ByteArrayInputStream("{\"cursor\":{\"partition\":\"0\",\"offset\":\"0\"}}".getBytes("utf-8")); when(response.getBody()).thenReturn(initialInputStream); final Request request = mock(Request.class); when(request.execute()).thenReturn(response).thenThrow(new IOException("Reconnection failed")).thenReturn(response).thenThrow(new IOException("Reconnection failed on second attempt")); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final ExponentialBackoffStrategy backoffStrategy = new EqualJitterBackoffStrategy(1, 1, 2, 4); final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener); expectedException.expect(BackoffException.class); expectedException.expect(ComposeMatchers.hasFeature(BackoffException::getRetries, equalTo(4))); expectedException.expectCause(instanceOf(IOException.class)); expectedException.expectCause(ComposeMatchers.hasFeature("message", Exception::getMessage, Matchers.containsString("Reconnection failed on second attempt"))); nakadiReader.runInternal(); }
@Test public void shouldProcessEventsAndCommitCursor() throws IOException, InterruptedException, BackoffException, EventAlreadyProcessedException { final Response response = mock(Response.class); String input = "{\"cursor\":{\"event_type\":\""+EVENT_NAME+"\",\"partition\":\"123\",\"offset\":\"456\"},\"events\":[{\"id\":\"789\"}]}"; final ByteArrayInputStream initialInputStream = new ByteArrayInputStream(input.getBytes("utf-8")); final ByteArrayInputStream emptyInputStream = new ByteArrayInputStream(new byte[0]); when(response.getBody()).thenReturn(initialInputStream, emptyInputStream); final Request request = mock(Request.class); when(request.execute()).thenReturn(response); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final NoBackoffStrategy backoffStrategy = new NoBackoffStrategy(); final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener); try { nakadiReader.runInternal(); fail("Expected IOException on reconnect"); } catch (BackoffException e) { assertEquals(0, e.getRetries()); assertEquals("Stream was closed", e.getCause().getMessage()); { @SuppressWarnings("unchecked") final ArgumentCaptor<List<SomeEvent>> argumentCaptor = (ArgumentCaptor<List<SomeEvent>>)(Object)ArgumentCaptor.forClass(List.class); verify(listener).accept(argumentCaptor.capture()); final List<SomeEvent> events = argumentCaptor.getValue(); assertEquals(1, events.size()); assertEquals("789", events.get(0).getId()); } { final ArgumentCaptor<Cursor> argumentCaptor = ArgumentCaptor.forClass(Cursor.class); verify(cursorManager).onSuccess(ArgumentMatchers.eq(EVENT_NAME), argumentCaptor.capture()); final Cursor cursor = argumentCaptor.getValue(); assertEquals("123", cursor.getPartition()); assertEquals("456", cursor.getOffset()); } } }
@Test public void shouldFailWithoutCursors() throws IOException, InterruptedException, BackoffException, EventAlreadyProcessedException { final Response response = mock(Response.class); final ByteArrayInputStream inputStream = new ByteArrayInputStream("{\"foo\":\"bar\"}".getBytes("utf-8")); when(response.getBody()).thenReturn(inputStream); final Request request = mock(Request.class); when(request.execute()).thenReturn(response); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final NoBackoffStrategy backoffStrategy = new NoBackoffStrategy(); final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener); expectedException.expect(BackoffException.class); expectedException.expect(ComposeMatchers.hasFeature(BackoffException::getRetries, equalTo(0))); expectedException.expectCause(instanceOf(IOException.class)); expectedException.expectCause(ComposeMatchers.hasFeature("message", Exception::getMessage, Matchers.containsString("Could not read cursor"))); nakadiReader.runInternal(); }
@Test public void shouldIgnoreMetadataInEventBatch() throws IOException, InterruptedException, BackoffException, EventAlreadyProcessedException { final Response response = mock(Response.class); final ByteArrayInputStream initialInputStream = new ByteArrayInputStream("{\"cursor\":{\"partition\":\"123\",\"offset\":\"456\"},\"events\":[{\"id\":\"789\"}],\"metadata\":{\"foo\":\"bar\"}}".getBytes("utf-8")); final ByteArrayInputStream emptyInputStream = new ByteArrayInputStream(new byte[0]); when(response.getBody()).thenReturn(initialInputStream, emptyInputStream); final Request request = mock(Request.class); when(request.execute()).thenReturn(response); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final NoBackoffStrategy backoffStrategy = new NoBackoffStrategy(); final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener); expectedException.expect(BackoffException.class); expectedException.expect(ComposeMatchers.hasFeature(BackoffException::getRetries, equalTo(0))); expectedException.expectCause(instanceOf(IOException.class)); expectedException.expectCause(ComposeMatchers.hasFeature("message", Exception::getMessage, Matchers.containsString("Stream was closed"))); nakadiReader.runInternal(); }
@Test public void shouldIgnoreAdditionalPropertiesInEventBatch() throws IOException, InterruptedException, BackoffException, EventAlreadyProcessedException { final Response response = mock(Response.class); final ByteArrayInputStream initialInputStream = new ByteArrayInputStream("{\"cursor\":{\"partition\":\"123\",\"offset\":\"456\"},\"foo\":\"bar\",\"events\":[{\"id\":\"789\"}],\"metadata\":{\"foo\":\"bar\"}}".getBytes("utf-8")); final ByteArrayInputStream emptyInputStream = new ByteArrayInputStream(new byte[0]); when(response.getBody()).thenReturn(initialInputStream, emptyInputStream); final Request request = mock(Request.class); when(request.execute()).thenReturn(response); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final NoBackoffStrategy backoffStrategy = new NoBackoffStrategy(); final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener); expectedException.expect(BackoffException.class); expectedException.expect(ComposeMatchers.hasFeature(BackoffException::getRetries, equalTo(0))); expectedException.expectCause(instanceOf(IOException.class)); expectedException.expectCause(ComposeMatchers.hasFeature("message", Exception::getMessage, Matchers.containsString("Stream was closed"))); nakadiReader.runInternal(); }
@Test public void shouldIgnoreAdditionalPropertiesInEventBatchInAnyOrder() throws IOException, InterruptedException, BackoffException, EventAlreadyProcessedException { final Response response = mock(Response.class); final ByteArrayInputStream initialInputStream = new ByteArrayInputStream("{\"foo\":\"bar\",\"cursor\":{\"partition\":\"123\",\"offset\":\"456\"},\"events\":[{\"id\":\"789\"}],\"baz\":123}".getBytes("utf-8")); final ByteArrayInputStream emptyInputStream = new ByteArrayInputStream(new byte[0]); when(response.getBody()).thenReturn(initialInputStream, emptyInputStream); final Request request = mock(Request.class); when(request.execute()).thenReturn(response); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final NoBackoffStrategy backoffStrategy = new NoBackoffStrategy(); final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener); expectedException.expect(BackoffException.class); expectedException.expect(ComposeMatchers.hasFeature(BackoffException::getRetries, equalTo(0))); expectedException.expectCause(instanceOf(IOException.class)); expectedException.expectCause(ComposeMatchers.hasFeature("message", Exception::getMessage, Matchers.containsString("Stream was closed"))); nakadiReader.runInternal(); }
@Test public void shouldExtractPropertyFromEvents() throws IOException, InterruptedException, BackoffException, EventAlreadyProcessedException { final Response response = mock(Response.class); final ByteArrayInputStream initialInputStream = new ByteArrayInputStream("{\"cursor\":{\"partition\":\"0\",\"offset\":\"0\"},\"events\":[{\"id\":\"1\",\"foo\":\"bar\"},{\"foo\":\"bar\",\"id\":\"2\"},{\"foo\":[\"bar\"],\"id\":\"3\",\"baz\":{\"id\":\"xyz\"}},{}]}".getBytes("utf-8")); final ByteArrayInputStream emptyInputStream = new ByteArrayInputStream(new byte[0]); when(response.getBody()).thenReturn(initialInputStream, emptyInputStream); final Request request = mock(Request.class); when(request.execute()).thenReturn(response); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final NoBackoffStrategy backoffStrategy = new NoBackoffStrategy(); final List<String> ids = new ArrayList<>(); final NakadiReader<String> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), new StringPropertyExtractingEventReader("id"), ids::addAll, DefaultBatchHandler.INSTANCE, NoMetricsCollector.NO_METRICS_COLLECTOR); expectedException.expect(BackoffException.class); expectedException.expect(ComposeMatchers.hasFeature(BackoffException::getRetries, equalTo(0))); expectedException.expectCause(instanceOf(IOException.class)); expectedException.expectCause(ComposeMatchers.hasFeature("message", Exception::getMessage, Matchers.containsString("Stream was closed"))); nakadiReader.runInternal(); assertEquals(asList("1", "2", "3"), ids); }
@Test public void shouldExtractPropertyFromEmptyEvents() throws IOException, InterruptedException, BackoffException, EventAlreadyProcessedException { final Response response = mock(Response.class); final ByteArrayInputStream initialInputStream = new ByteArrayInputStream("{\"cursor\":{\"partition\":\"0\",\"offset\":\"0\"},\"events\":[]}".getBytes("utf-8")); final ByteArrayInputStream emptyInputStream = new ByteArrayInputStream(new byte[0]); when(response.getBody()).thenReturn(initialInputStream, emptyInputStream); final Request request = mock(Request.class); when(request.execute()).thenReturn(response); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final NoBackoffStrategy backoffStrategy = new NoBackoffStrategy(); final List<String> ids = new ArrayList<>(); final NakadiReader<String> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), new StringPropertyExtractingEventReader("id"), ids::addAll, DefaultBatchHandler.INSTANCE, NoMetricsCollector.NO_METRICS_COLLECTOR); expectedException.expect(BackoffException.class); expectedException.expect(ComposeMatchers.hasFeature(BackoffException::getRetries, equalTo(0))); expectedException.expectCause(instanceOf(IOException.class)); expectedException.expectCause(ComposeMatchers.hasFeature("message", Exception::getMessage, Matchers.containsString("Stream was closed"))); nakadiReader.runInternal(); assertEquals(emptyList(), ids); }