luciano/amazon-kinesis-client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Prevent improper error logging during worker shutdown (#1257)

Browse source 
* Move throwOnIllegalState call to drain queue method to prevent improper error logging during worker shutdown

* Fix unit tests that expected IllegalStateException thrown

* Changed names of unit tests to reflect new behavior
This commit is contained in:
zachjhum 2024-02-21 13:19:50 -08:00 committed by GitHub
parent b2eb38e510
commit 5f3de14c88
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 23 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
amazon-kinesis-client/src/main/java/software/amazon/kinesis/retrieval/polling/PrefetchRecordsPublisher.java
View file

@ -289,7 +289,6 @@ public class PrefetchRecordsPublisher implements RecordsPublisher {
} }
private PrefetchRecordsRetrieved peekNextResult() { private PrefetchRecordsRetrieved peekNextResult() {
throwOnIllegalState();
return publisherSession.peekNextRecord(); return publisherSession.peekNextRecord();
} }
@ -336,6 +335,7 @@ public class PrefetchRecordsPublisher implements RecordsPublisher {
@Override @Override
public void subscribe(Subscriber<? super RecordsRetrieved> s) { public void subscribe(Subscriber<? super RecordsRetrieved> s) {
throwOnIllegalState();
subscriber = s; subscriber = s;
subscriber.onSubscribe(new Subscription() { subscriber.onSubscribe(new Subscription() {
@Override @Override
@ -389,6 +389,7 @@ public class PrefetchRecordsPublisher implements RecordsPublisher {
// If there is an event available to drain and if there is at least one demand, // If there is an event available to drain and if there is at least one demand,
// then schedule it for delivery // then schedule it for delivery
if (publisherSession.hasDemandToPublish() && canDispatchRecord(recordsToDeliver)) { if (publisherSession.hasDemandToPublish() && canDispatchRecord(recordsToDeliver)) {
throwOnIllegalState();
subscriber.onNext(recordsToDeliver.prepareForPublish()); subscriber.onNext(recordsToDeliver.prepareForPublish());
recordsToDeliver.dispatched(); recordsToDeliver.dispatched();
lastEventDeliveryTime = Instant.now(); lastEventDeliveryTime = Instant.now();

25
amazon-kinesis-client/src/test/java/software/amazon/kinesis/retrieval/polling/PrefetchRecordsPublisherTest.java
View file

@ -31,8 +31,10 @@ import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing; import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never; import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy; import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.times; import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when; import static org.mockito.Mockito.when;
@ -64,6 +66,7 @@ import org.junit.Before;
import org.junit.Ignore; import org.junit.Ignore;
import org.junit.Test; import org.junit.Test;
import org.junit.runner.RunWith; import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock; import org.mockito.Mock;
import org.mockito.invocation.InvocationOnMock; import org.mockito.invocation.InvocationOnMock;
import org.mockito.runners.MockitoJUnitRunner; import org.mockito.runners.MockitoJUnitRunner;
@ -375,15 +378,29 @@ public class PrefetchRecordsPublisherTest {
} }
@Test(expected = IllegalStateException.class) @Test(expected = IllegalStateException.class)
public void testGetNextRecordsWithoutStarting() { public void testSubscribeWithoutStarting() {
verify(executorService, never()).execute(any()); verify(executorService, never()).execute(any());
getRecordsCache.drainQueueForRequests(); Subscriber<RecordsRetrieved> mockSubscriber = mock(Subscriber.class);
getRecordsCache.subscribe(mockSubscriber);
} }
@Test(expected = IllegalStateException.class) @Test(expected = IllegalStateException.class)
public void testCallAfterShutdown() { public void testRequestRecordsOnSubscriptionAfterShutdown() {
GetRecordsResponse response = GetRecordsResponse.builder().records(
Record.builder().data(SdkBytes.fromByteArray(new byte[] { 1, 2, 3 })).sequenceNumber("123").build())
.nextShardIterator(NEXT_SHARD_ITERATOR).build();
when(getRecordsRetrievalStrategy.getRecords(anyInt())).thenReturn(response);
getRecordsCache.start(sequenceNumber, initialPosition);
verify(getRecordsRetrievalStrategy, timeout(100).atLeastOnce()).getRecords(anyInt());
when(executorService.isShutdown()).thenReturn(true); when(executorService.isShutdown()).thenReturn(true);
getRecordsCache.drainQueueForRequests(); Subscriber<RecordsRetrieved> mockSubscriber = mock(Subscriber.class);
getRecordsCache.subscribe(mockSubscriber);
ArgumentCaptor<Subscription> subscriptionCaptor = ArgumentCaptor.forClass(Subscription.class);
verify(mockSubscriber).onSubscribe(subscriptionCaptor.capture());
subscriptionCaptor.getValue().request(1);
} }
@Test @Test