@Test public void loggingNotSuppressedOnNonReadTimeoutExceptionIgnoringReadTimeoutsTest() { Exception exceptionToThrow = new RuntimeException("Uh oh Not a ReadTimeout"); int readTimeoutsToIgnore = 2; int expectedReadTimeoutLogs = 0; int expectedGenericLogs = 2; TestShardConsumerSubscriber consumer = new TestShardConsumerSubscriber(mock(RecordsPublisher.class), Executors.newFixedThreadPool(1), 8, shardConsumer, readTimeoutsToIgnore); consumer.startSubscriptions(); mimicSuccess(consumer); mimicSuccess(consumer); mimicException(exceptionToThrow, consumer); mimicSuccess(consumer); mimicException(exceptionToThrow, consumer); assertEquals(expectedGenericLogs, consumer.genericWarningLogged); assertEquals(expectedReadTimeoutLogs, consumer.readTimeoutWarningLogged); }
void startSubscriptions() { synchronized (lockObject) { lastRequestTime = Instant.now(); if (lastAccepted != null) { recordsPublisher.restartFrom(lastAccepted); } Flowable.fromPublisher(recordsPublisher).subscribeOn(scheduler).observeOn(scheduler, true, bufferSize) .subscribe(new ShardConsumerNotifyingSubscriber(this, recordsPublisher)); } }
ShardConsumerSubscriber implements Subscriber<RecordsRetrieved> { void startSubscriptions() { synchronized (lockObject) { lastRequestTime = Instant.now(); if (lastAccepted != null) { recordsPublisher.restartFrom(lastAccepted); } Flowable.fromPublisher(recordsPublisher).subscribeOn(scheduler).observeOn(scheduler, true, bufferSize) .subscribe(new ShardConsumerNotifyingSubscriber(this, recordsPublisher)); } } }
ShardConsumerSubscriber implements Subscriber<RecordsRetrieved> { void startSubscriptions() { synchronized (lockObject) { lastRequestTime = Instant.now(); if (lastAccepted != null) { recordsPublisher.restartFrom(lastAccepted); } Flowable.fromPublisher(recordsPublisher).subscribeOn(scheduler).observeOn(scheduler, true, bufferSize) .subscribe(new ShardConsumerNotifyingSubscriber(this, recordsPublisher)); } } @Deprecated  ShardConsumerSubscriber(RecordsPublisher recordsPublisher, ExecutorService executorService, int bufferSize,
                            ShardConsumer shardConsumer);   ShardConsumerSubscriber(RecordsPublisher recordsPublisher, ExecutorService executorService, int bufferSize,
            ShardConsumer shardConsumer, int readTimeoutsToIgnoreBeforeWarning); }
ShardConsumerSubscriber implements Subscriber<RecordsRetrieved> { void startSubscriptions() { synchronized (lockObject) { lastRequestTime = Instant.now(); if (lastAccepted != null) { recordsPublisher.restartFrom(lastAccepted); } Flowable.fromPublisher(recordsPublisher).subscribeOn(scheduler).observeOn(scheduler, true, bufferSize) .subscribe(new ShardConsumerNotifyingSubscriber(this, recordsPublisher)); } } @Deprecated  ShardConsumerSubscriber(RecordsPublisher recordsPublisher, ExecutorService executorService, int bufferSize,
                            ShardConsumer shardConsumer);   ShardConsumerSubscriber(RecordsPublisher recordsPublisher, ExecutorService executorService, int bufferSize,
            ShardConsumer shardConsumer, int readTimeoutsToIgnoreBeforeWarning); @Override void onSubscribe(Subscription s); @Override void onNext(RecordsRetrieved input); @Override void onError(Throwable t); @Override void onComplete(); void cancel(); }
ShardConsumerSubscriber implements Subscriber<RecordsRetrieved> { void startSubscriptions() { synchronized (lockObject) { lastRequestTime = Instant.now(); if (lastAccepted != null) { recordsPublisher.restartFrom(lastAccepted); } Flowable.fromPublisher(recordsPublisher).subscribeOn(scheduler).observeOn(scheduler, true, bufferSize) .subscribe(new ShardConsumerNotifyingSubscriber(this, recordsPublisher)); } } @Deprecated  ShardConsumerSubscriber(RecordsPublisher recordsPublisher, ExecutorService executorService, int bufferSize,
                            ShardConsumer shardConsumer);   ShardConsumerSubscriber(RecordsPublisher recordsPublisher, ExecutorService executorService, int bufferSize,
            ShardConsumer shardConsumer, int readTimeoutsToIgnoreBeforeWarning); @Override void onSubscribe(Subscription s); @Override void onNext(RecordsRetrieved input); @Override void onError(Throwable t); @Override void onComplete(); void cancel();  }