diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventProcessorClient.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventProcessorClient.java index 044c6f3de373..52104c1184fd 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventProcessorClient.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventProcessorClient.java @@ -7,19 +7,19 @@ import com.azure.core.annotation.ServiceClient; import com.azure.core.util.logging.ClientLogger; import com.azure.messaging.eventhubs.implementation.PartitionProcessor; -import com.azure.messaging.eventhubs.models.EventPosition; +import com.azure.messaging.eventhubs.models.ErrorContext; import java.util.Locale; -import java.util.concurrent.atomic.AtomicReference; -import reactor.core.Disposable; -import reactor.core.scheduler.Scheduler; -import reactor.core.scheduler.Schedulers; - import java.util.Objects; import java.util.UUID; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; import java.util.function.Supplier; +import reactor.core.Disposable; +import reactor.core.scheduler.Scheduler; +import reactor.core.scheduler.Schedulers; /** * EventProcessorClient provides a convenient mechanism to consume events from all partitions of an Event Hub in the @@ -54,33 +54,31 @@ public class EventProcessorClient { * @param eventHubClientBuilder The {@link EventHubClientBuilder}. * @param consumerGroup The consumer group name used in this event processor to consumer events. * @param partitionProcessorFactory The factory to create new partition processor(s). - * @param initialEventPosition Initial event position to start consuming events. * @param checkpointStore The store used for reading and updating partition ownership and checkpoints. information. * @param trackLastEnqueuedEventProperties If set to {@code true}, all events received by this * EventProcessorClient will also include the last enqueued event properties for it's respective partitions. * @param tracerProvider The tracer implementation. */ EventProcessorClient(EventHubClientBuilder eventHubClientBuilder, String consumerGroup, - Supplier partitionProcessorFactory, EventPosition initialEventPosition, - CheckpointStore checkpointStore, boolean trackLastEnqueuedEventProperties, TracerProvider tracerProvider) { + Supplier partitionProcessorFactory, CheckpointStore checkpointStore, + boolean trackLastEnqueuedEventProperties, TracerProvider tracerProvider, Consumer processError) { Objects.requireNonNull(eventHubClientBuilder, "eventHubClientBuilder cannot be null."); Objects.requireNonNull(consumerGroup, "consumerGroup cannot be null."); Objects.requireNonNull(partitionProcessorFactory, "partitionProcessorFactory cannot be null."); - Objects.requireNonNull(initialEventPosition, "initialEventPosition cannot be null."); this.checkpointStore = Objects.requireNonNull(checkpointStore, "checkpointStore cannot be null"); this.identifier = UUID.randomUUID().toString(); logger.info("The instance ID for this event processors is {}", this.identifier); this.partitionPumpManager = new PartitionPumpManager(checkpointStore, partitionProcessorFactory, - initialEventPosition, eventHubClientBuilder, trackLastEnqueuedEventProperties, tracerProvider); + eventHubClientBuilder, trackLastEnqueuedEventProperties, tracerProvider); EventHubAsyncClient eventHubAsyncClient = eventHubClientBuilder.buildAsyncClient(); this.partitionBasedLoadBalancer = new PartitionBasedLoadBalancer(this.checkpointStore, eventHubAsyncClient, eventHubAsyncClient.getFullyQualifiedNamespace().toLowerCase(Locale.ROOT), eventHubAsyncClient.getEventHubName().toLowerCase(Locale.ROOT), consumerGroup.toLowerCase(Locale.ROOT), identifier, TimeUnit.MINUTES.toSeconds(1), - partitionPumpManager); + partitionPumpManager, processError); } /** diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventProcessorClientBuilder.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventProcessorClientBuilder.java index aef2a3a63100..daafe27c3d68 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventProcessorClientBuilder.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventProcessorClientBuilder.java @@ -321,8 +321,8 @@ public EventProcessorClient buildEventProcessorClient() { final TracerProvider tracerProvider = new TracerProvider(ServiceLoader.load(Tracer.class)); return new EventProcessorClient(eventHubClientBuilder, this.consumerGroup, - getPartitionProcessorSupplier(), EventPosition.earliest(), checkpointStore, - trackLastEnqueuedEventProperties, tracerProvider); + getPartitionProcessorSupplier(), checkpointStore, trackLastEnqueuedEventProperties, tracerProvider, + processError); } private Supplier getPartitionProcessorSupplier() { diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/PartitionBasedLoadBalancer.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/PartitionBasedLoadBalancer.java index e72e5b932b8b..5a89d6d498f0 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/PartitionBasedLoadBalancer.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/PartitionBasedLoadBalancer.java @@ -5,7 +5,10 @@ import com.azure.core.util.CoreUtils; import com.azure.core.util.logging.ClientLogger; +import com.azure.messaging.eventhubs.models.ErrorContext; +import com.azure.messaging.eventhubs.models.PartitionContext; import com.azure.messaging.eventhubs.models.PartitionOwnership; +import java.util.function.Consumer; import reactor.core.Exceptions; import reactor.core.publisher.Mono; import reactor.util.function.Tuple2; @@ -47,6 +50,8 @@ final class PartitionBasedLoadBalancer { private final long inactiveTimeLimitInSeconds; private final PartitionPumpManager partitionPumpManager; private final String fullyQualifiedNamespace; + private final Consumer processError; + private final PartitionContext partitionAgnosticContext; /** * Creates an instance of PartitionBasedLoadBalancer for the given Event Hub name and consumer group. @@ -60,11 +65,13 @@ final class PartitionBasedLoadBalancer { * assuming the owner of the partition is inactive. * @param partitionPumpManager The partition pump manager that keeps track of all EventHubConsumers and partitions * that this {@link EventProcessorClient} is processing. + * @param processError The callback that will be called when an error occurs while running the load balancer. */ PartitionBasedLoadBalancer(final CheckpointStore checkpointStore, final EventHubAsyncClient eventHubAsyncClient, final String fullyQualifiedNamespace, final String eventHubName, final String consumerGroupName, final String ownerId, - final long inactiveTimeLimitInSeconds, final PartitionPumpManager partitionPumpManager) { + final long inactiveTimeLimitInSeconds, final PartitionPumpManager partitionPumpManager, + final Consumer processError) { this.checkpointStore = checkpointStore; this.eventHubAsyncClient = eventHubAsyncClient; this.fullyQualifiedNamespace = fullyQualifiedNamespace; @@ -73,6 +80,9 @@ final class PartitionBasedLoadBalancer { this.ownerId = ownerId; this.inactiveTimeLimitInSeconds = inactiveTimeLimitInSeconds; this.partitionPumpManager = partitionPumpManager; + this.processError = processError; + this.partitionAgnosticContext = new PartitionContext(fullyQualifiedNamespace, eventHubName, + consumerGroupName, "NONE"); } /** @@ -107,8 +117,12 @@ void loadBalance() { Mono.zip(partitionOwnershipMono, partitionsMono) .flatMap(this::loadBalance) // if there was an error, log warning and TODO: call user provided error handler - .doOnError(ex -> logger.warning(Messages.LOAD_BALANCING_FAILED, ex.getMessage())) - .subscribe(); + .subscribe(ignored -> { }, + ex -> { + logger.warning(Messages.LOAD_BALANCING_FAILED, ex.getMessage()); + ErrorContext errorContext = new ErrorContext(partitionAgnosticContext, ex); + processError.accept(errorContext); + }, () -> logger.info("Load balancing completed successfully")); } /* @@ -350,7 +364,10 @@ private void claimOwnership(final Map partitionOwner .stream() .forEach(po -> partitionPumpManager.startPartitionPump(po, ownedPartitionCheckpointsTuple.getT2().get(po.getPartitionId()))); - }); + }, + ex -> { + throw logger.logExceptionAsError(new RuntimeException("Error while listing checkpoints", ex)); + }); } private PartitionOwnership createPartitionOwnershipRequest( diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/PartitionPumpManager.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/PartitionPumpManager.java index 451485fd573b..a620f1392c2e 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/PartitionPumpManager.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/PartitionPumpManager.java @@ -3,7 +3,12 @@ package com.azure.messaging.eventhubs; -import com.azure.core.amqp.exception.AmqpException; +import static com.azure.core.util.tracing.Tracer.DIAGNOSTIC_ID_KEY; +import static com.azure.core.util.tracing.Tracer.SCOPE_KEY; +import static com.azure.core.util.tracing.Tracer.SPAN_CONTEXT_KEY; +import static com.azure.core.util.tracing.Tracer.ENTITY_PATH_KEY; +import static com.azure.core.util.tracing.Tracer.HOST_NAME_KEY; + import com.azure.core.amqp.implementation.TracerProvider; import com.azure.core.util.Context; import com.azure.core.util.logging.ClientLogger; @@ -12,15 +17,13 @@ import com.azure.messaging.eventhubs.models.Checkpoint; import com.azure.messaging.eventhubs.models.CloseContext; import com.azure.messaging.eventhubs.models.CloseReason; +import com.azure.messaging.eventhubs.models.ErrorContext; import com.azure.messaging.eventhubs.models.EventContext; import com.azure.messaging.eventhubs.models.EventPosition; -import com.azure.messaging.eventhubs.models.ErrorContext; import com.azure.messaging.eventhubs.models.InitializationContext; import com.azure.messaging.eventhubs.models.PartitionContext; import com.azure.messaging.eventhubs.models.PartitionOwnership; import com.azure.messaging.eventhubs.models.ReceiveOptions; -import reactor.core.publisher.Signal; - import java.io.Closeable; import java.io.IOException; import java.util.Locale; @@ -28,12 +31,7 @@ import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.function.Supplier; - -import static com.azure.core.util.tracing.Tracer.DIAGNOSTIC_ID_KEY; -import static com.azure.core.util.tracing.Tracer.ENTITY_PATH_KEY; -import static com.azure.core.util.tracing.Tracer.HOST_NAME_KEY; -import static com.azure.core.util.tracing.Tracer.SCOPE_KEY; -import static com.azure.core.util.tracing.Tracer.SPAN_CONTEXT_KEY; +import reactor.core.publisher.Signal; /** * The partition pump manager that keeps track of all the partition pumps started by this {@link EventProcessorClient}. @@ -52,7 +50,6 @@ class PartitionPumpManager { private final CheckpointStore checkpointStore; private final Map partitionPumps = new ConcurrentHashMap<>(); private final Supplier partitionProcessorFactory; - private final EventPosition initialEventPosition; private final EventHubClientBuilder eventHubClientBuilder; private final TracerProvider tracerProvider; private final boolean trackLastEnqueuedEventProperties; @@ -63,21 +60,17 @@ class PartitionPumpManager { * @param checkpointStore The partition manager that is used to store and update checkpoints. * @param partitionProcessorFactory The partition processor factory that is used to create new instances of {@link * PartitionProcessor} when new partition pumps are started. - * @param initialEventPosition The initial event position to use when a new partition pump is created and no - * checkpoint for the partition is available. * @param eventHubClientBuilder The client builder used to create new clients (and new connections) for each * partition processed by this {@link EventProcessorClient}. - * @param trackLastEnqueuedEventProperties If set to {@code true}, all events received by this - * EventProcessorClient will also include the last enqueued event properties for it's respective partitions. + * @param trackLastEnqueuedEventProperties If set to {@code true}, all events received by this EventProcessorClient + * will also include the last enqueued event properties for it's respective partitions. * @param tracerProvider The tracer implementation. */ PartitionPumpManager(CheckpointStore checkpointStore, - Supplier partitionProcessorFactory, EventPosition initialEventPosition, - EventHubClientBuilder eventHubClientBuilder, boolean trackLastEnqueuedEventProperties, - TracerProvider tracerProvider) { + Supplier partitionProcessorFactory, EventHubClientBuilder eventHubClientBuilder, + boolean trackLastEnqueuedEventProperties, TracerProvider tracerProvider) { this.checkpointStore = checkpointStore; this.partitionProcessorFactory = partitionProcessorFactory; - this.initialEventPosition = initialEventPosition; this.eventHubClientBuilder = eventHubClientBuilder; this.trackLastEnqueuedEventProperties = trackLastEnqueuedEventProperties; this.tracerProvider = tracerProvider; @@ -121,12 +114,17 @@ void startPartitionPump(PartitionOwnership claimedOwnership, Checkpoint checkpoi partitionProcessor.initialize(initializationContext); EventPosition startFromEventPosition = null; + // A checkpoint indicates the last known successfully processed event. + // So, the event position to start a new partition processing should be exclusive of the + // offset/sequence number in the checkpoint. If no checkpoint is available, start from + // the position in set in the InitializationContext (either the earliest event in the partition or + // the user provided initial position) if (checkpoint != null && checkpoint.getOffset() != null) { startFromEventPosition = EventPosition.fromOffset(checkpoint.getOffset()); } else if (checkpoint != null && checkpoint.getSequenceNumber() != null) { - startFromEventPosition = EventPosition.fromSequenceNumber(checkpoint.getSequenceNumber(), true); + startFromEventPosition = EventPosition.fromSequenceNumber(checkpoint.getSequenceNumber()); } else { - startFromEventPosition = initialEventPosition; + startFromEventPosition = initializationContext.getInitialPosition(); } ReceiveOptions receiveOptions = new ReceiveOptions().setOwnerLevel(0L) @@ -151,42 +149,27 @@ void startPartitionPump(PartitionOwnership claimedOwnership, Checkpoint checkpoi partitionProcessor.processEvent(new EventContext(partitionContext, eventData, checkpointStore, partitionEvent.getLastEnqueuedEventProperties())); endProcessTracingSpan(processSpanContext, Signal.complete()); - } catch (Exception ex) { - /* event processing threw an exception */ - handleProcessingError(claimedOwnership, partitionProcessor, ex, partitionContext); - endProcessTracingSpan(processSpanContext, Signal.error(ex)); - + } catch (Throwable throwable) { + /* user code for event processing threw an exception - log and bubble up */ + endProcessTracingSpan(processSpanContext, Signal.error(throwable)); + throw logger.logExceptionAsError(new RuntimeException("Error in event processing callback", + throwable)); } }, /* EventHubConsumer receive() returned an error */ ex -> handleReceiveError(claimedOwnership, eventHubConsumer, partitionProcessor, ex, partitionContext), () -> partitionProcessor.close(new CloseContext(partitionContext, CloseReason.EVENT_PROCESSOR_SHUTDOWN))); - // @formatter:on - } - - private void handleProcessingError(PartitionOwnership claimedOwnership, PartitionProcessor partitionProcessor, - Throwable error, PartitionContext partitionContext) { - try { - // There was an error in process event (user provided code), call process error and if that - // also fails just log and continue - partitionProcessor.processError(new ErrorContext(partitionContext, error)); - } catch (Exception ex) { - logger.warning(Messages.FAILED_WHILE_PROCESSING_ERROR, claimedOwnership.getPartitionId(), ex); - } + // @formatter:on } private void handleReceiveError(PartitionOwnership claimedOwnership, EventHubConsumerAsyncClient eventHubConsumer, - PartitionProcessor partitionProcessor, Throwable error, PartitionContext partitionContext) { + PartitionProcessor partitionProcessor, Throwable throwable, PartitionContext partitionContext) { try { + logger.warning("Error receiving events for partition {}", partitionContext.getPartitionId(), throwable); // if there was an error on receive, it also marks the end of the event data stream - partitionProcessor.processError(new ErrorContext(partitionContext, error)); - CloseReason closeReason = CloseReason.EVENT_HUB_EXCEPTION; - // If the exception indicates that the partition was stolen (i.e some other consumer with same ownerlevel - // started consuming the partition), update the closeReason - // TODO: Find right exception type to determine stolen partition - if (error instanceof AmqpException) { - closeReason = CloseReason.LOST_PARTITION_OWNERSHIP; - } + partitionProcessor.processError(new ErrorContext(partitionContext, throwable)); + // Any exception while receiving events will result in the processor losing ownership + CloseReason closeReason = CloseReason.LOST_PARTITION_OWNERSHIP; partitionProcessor.close(new CloseContext(partitionContext, closeReason)); } catch (Exception ex) { logger.warning(Messages.FAILED_PROCESSING_ERROR_RECEIVE, claimedOwnership.getPartitionId(), ex); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/CloseReason.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/CloseReason.java index 09d591745b9a..0d5a39622330 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/CloseReason.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/CloseReason.java @@ -21,10 +21,4 @@ public enum CloseReason { * PartitionProcessor#close(CloseContext)} will be called with this reason. */ EVENT_PROCESSOR_SHUTDOWN, - - /** - * If a non-retryable exception occured when receiving events from Event Hub, this reason will be provided when - * {@link PartitionProcessor#close(CloseContext)} is called. - */ - EVENT_HUB_EXCEPTION } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventProcessorClientErrorHandlingTest.java b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventProcessorClientErrorHandlingTest.java new file mode 100644 index 000000000000..21b799801c8b --- /dev/null +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventProcessorClientErrorHandlingTest.java @@ -0,0 +1,194 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.messaging.eventhubs; + +import static org.mockito.Mockito.when; + +import com.azure.messaging.eventhubs.implementation.ClientConstants; +import com.azure.messaging.eventhubs.implementation.PartitionProcessor; +import com.azure.messaging.eventhubs.models.Checkpoint; +import com.azure.messaging.eventhubs.models.ErrorContext; +import com.azure.messaging.eventhubs.models.EventContext; +import com.azure.messaging.eventhubs.models.PartitionOwnership; +import java.net.URI; +import java.net.URISyntaxException; +import java.util.List; +import java.util.Locale; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.stream.Stream; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * Unit tests for {@link EventProcessorClient} error handling. + */ +public class EventProcessorClientErrorHandlingTest { + + private static final String NAMESPACE_NAME = "dummyNamespaceName"; + private static final String DEFAULT_DOMAIN_NAME = "servicebus.windows.net/"; + private static final String EVENT_HUB_NAME = "eventHubName"; + private static final String SHARED_ACCESS_KEY_NAME = "dummySasKeyName"; + private static final String SHARED_ACCESS_KEY = "dummySasKey"; + private static final String ENDPOINT = getURI(ClientConstants.ENDPOINT_FORMAT, NAMESPACE_NAME, DEFAULT_DOMAIN_NAME) + .toString(); + + private static final String CORRECT_CONNECTION_STRING = String + .format("Endpoint=%s;SharedAccessKeyName=%s;SharedAccessKey=%s;EntityPath=%s", + ENDPOINT, SHARED_ACCESS_KEY_NAME, SHARED_ACCESS_KEY, EVENT_HUB_NAME); + + @Mock + private EventHubClientBuilder eventHubClientBuilder; + + @Mock + private EventHubAsyncClient eventHubAsyncClient; + + @Mock + private EventHubConsumerAsyncClient eventHubConsumer; + + private CountDownLatch countDownLatch; + + private static URI getURI(String endpointFormat, String namespace, String domainName) { + try { + return new URI(String.format(Locale.US, endpointFormat, namespace, domainName)); + } catch (URISyntaxException exception) { + throw new IllegalArgumentException(String.format(Locale.US, + "Invalid namespace name: %s", namespace), exception); + } + } + + @BeforeEach + public void setup() { + MockitoAnnotations.initMocks(this); + when(eventHubClientBuilder.buildAsyncClient()).thenReturn(eventHubAsyncClient); + when(eventHubAsyncClient.getPartitionIds()).thenReturn(Flux.just("1", "2", "3")); + when(eventHubAsyncClient.getFullyQualifiedNamespace()).thenReturn("test-ns"); + when(eventHubAsyncClient.getEventHubName()).thenReturn("test-eh"); + } + + @ParameterizedTest(name = "{displayName} with [{arguments}]") + @MethodSource("checkpointStoreSupplier") + public void testCheckpointStoreErrors(CheckpointStore checkpointStore) throws InterruptedException { + countDownLatch = new CountDownLatch(1); + EventProcessorClient client = new EventProcessorClient(eventHubClientBuilder, "cg", + () -> new TestPartitionProcessor(), checkpointStore, false, + null, errorContext -> { + countDownLatch.countDown(); + Assertions.assertEquals("NONE", errorContext.getPartitionContext().getPartitionId()); + Assertions.assertEquals("cg", errorContext.getPartitionContext().getConsumerGroup()); + Assertions.assertTrue(errorContext.getThrowable() instanceof IllegalStateException); + }); + client.start(); + boolean completed = countDownLatch.await(3, TimeUnit.SECONDS); + client.stop(); + Assertions.assertTrue(completed); + } + + private static Stream checkpointStoreSupplier() { + return Stream.of( + Arguments.of(new ListOwnershipErrorStore()), + Arguments.of(new ClaimOwnershipErrorStore()), + Arguments.of(new ListCheckpointErrorStore())); + } + + + private static class ListOwnershipErrorStore implements CheckpointStore { + + @Override + public Flux listOwnership( + String fullyQualifiedNamespace, String eventHubName, String consumerGroup) { + return Flux.error(new IllegalStateException("List ownership error")); + } + + @Override + public Flux claimOwnership( + List requestedPartitionOwnerships) { + return null; + } + + @Override + public Flux listCheckpoints(String fullyQualifiedNamespace, + String eventHubName, String consumerGroup) { + return null; + } + + @Override + public Mono updateCheckpoint(Checkpoint checkpoint) { + return null; + } + } + + private static class ClaimOwnershipErrorStore implements CheckpointStore { + + @Override + public Flux listOwnership( + String fullyQualifiedNamespace, String eventHubName, String consumerGroup) { + return Flux.empty(); + } + + @Override + public Flux claimOwnership( + List requestedPartitionOwnerships) { + return Flux.error(new IllegalStateException("Claim Ownership error")); + } + + @Override + public Flux listCheckpoints(String fullyQualifiedNamespace, + String eventHubName, String consumerGroup) { + return null; + } + + @Override + public Mono updateCheckpoint(Checkpoint checkpoint) { + return null; + } + } + + private static class ListCheckpointErrorStore implements CheckpointStore { + + @Override + public Flux listOwnership( + String fullyQualifiedNamespace, String eventHubName, String consumerGroup) { + return Flux.empty(); + } + + @Override + public Flux claimOwnership( + List requestedPartitionOwnerships) { + return Flux.fromIterable(requestedPartitionOwnerships); + } + + @Override + public Flux listCheckpoints(String fullyQualifiedNamespace, + String eventHubName, String consumerGroup) { + return Flux.error(new IllegalStateException("List checkpoint error")); + } + + @Override + public Mono updateCheckpoint(Checkpoint checkpoint) { + return null; + } + } + + private static final class TestPartitionProcessor extends PartitionProcessor { + + @Override + public void processEvent(EventContext eventContext) { + eventContext.updateCheckpoint(); + } + + @Override + public void processError(ErrorContext errorContext) { + // do nothing + return; + } + } +} diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventProcessorClientTest.java b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventProcessorClientTest.java index d21a3d870b4e..7f1f47ad825b 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventProcessorClientTest.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventProcessorClientTest.java @@ -140,7 +140,7 @@ public void testWithSimplePartitionProcessor() throws Exception { // Act final EventProcessorClient eventProcessorClient = new EventProcessorClient(eventHubClientBuilder, "test-consumer", - () -> testPartitionProcessor, EventPosition.earliest(), checkpointStore, false, tracerProvider); + () -> testPartitionProcessor, checkpointStore, false, tracerProvider, ec -> { }); eventProcessorClient.start(); TimeUnit.SECONDS.sleep(10); eventProcessorClient.stop(); @@ -211,7 +211,7 @@ public void testWithFaultyPartitionProcessor() throws Exception { ); // Act final EventProcessorClient eventProcessorClient = new EventProcessorClient(eventHubClientBuilder, "test-consumer", - () -> faultyPartitionProcessor, EventPosition.earliest(), checkpointStore, false, tracerProvider); + () -> faultyPartitionProcessor, checkpointStore, false, tracerProvider, ec -> { }); eventProcessorClient.start(); TimeUnit.SECONDS.sleep(10); @@ -270,7 +270,7 @@ public void testErrorProcessSpans() throws Exception { //Act final EventProcessorClient eventProcessorClient = new EventProcessorClient(eventHubClientBuilder, "test-consumer", - FaultyPartitionProcessor::new, EventPosition.earliest(), checkpointStore, false, tracerProvider); + FaultyPartitionProcessor::new, checkpointStore, false, tracerProvider, ec -> { }); eventProcessorClient.start(); TimeUnit.SECONDS.sleep(10); eventProcessorClient.stop(); @@ -329,7 +329,7 @@ public void testProcessSpans() throws Exception { //Act final EventProcessorClient eventProcessorClient = new EventProcessorClient(eventHubClientBuilder, "test-consumer", - TestPartitionProcessor::new, EventPosition.earliest(), checkpointStore, false, tracerProvider); + TestPartitionProcessor::new, checkpointStore, false, tracerProvider, ec -> { }); eventProcessorClient.start(); TimeUnit.SECONDS.sleep(10); @@ -354,7 +354,7 @@ public void testWithMultiplePartitions() throws Exception { identifiers.add("1"); identifiers.add("2"); identifiers.add("3"); - final EventPosition position = EventPosition.latest(); + final EventPosition position = EventPosition.earliest(); when(eventHubClientBuilder.buildAsyncClient()).thenReturn(eventHubAsyncClient); when(eventHubAsyncClient.getPartitionIds()).thenReturn(Flux.just("1", "2", "3")); @@ -391,7 +391,7 @@ public void testWithMultiplePartitions() throws Exception { // Act final EventProcessorClient eventProcessorClient = new EventProcessorClient(eventHubClientBuilder, "test-consumer", - TestPartitionProcessor::new, position, checkpointStore, false, tracerProvider); + TestPartitionProcessor::new, checkpointStore, false, tracerProvider, ec -> { }); eventProcessorClient.start(); final boolean completed = count.await(10, TimeUnit.SECONDS); eventProcessorClient.stop(); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/PartitionBasedLoadBalancerTest.java b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/PartitionBasedLoadBalancerTest.java index eaeb4d322e32..6947ceccb41c 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/PartitionBasedLoadBalancerTest.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/PartitionBasedLoadBalancerTest.java @@ -312,11 +312,11 @@ public void testReceiveFailure() throws Exception { .thenReturn(Flux.error(new IllegalStateException())); PartitionPumpManager partitionPumpManager = new PartitionPumpManager(checkpointStore, - () -> partitionProcessor, EventPosition.earliest(), - eventHubClientBuilder, false, tracerProvider); + () -> partitionProcessor, eventHubClientBuilder, false, tracerProvider); PartitionBasedLoadBalancer loadBalancer = new PartitionBasedLoadBalancer(checkpointStore, eventHubAsyncClient, fqNamespace, eventHubName, consumerGroupName, "owner", TimeUnit.SECONDS.toSeconds(5), - partitionPumpManager); + partitionPumpManager, ec -> { + }); loadBalancer.loadBalance(); sleep(2); verify(partitionProcessor, never()).processEvent(any(EventContext.class)); @@ -337,10 +337,11 @@ public void testCheckpointStoreFailure() throws Exception { List partitionIds = Arrays.asList("1", "2", "3"); when(eventHubAsyncClient.getPartitionIds()).thenReturn(Flux.fromIterable(partitionIds)); PartitionPumpManager partitionPumpManager = new PartitionPumpManager(checkpointStore, - () -> partitionProcessor, EventPosition.earliest(), eventHubClientBuilder, false, tracerProvider); + () -> partitionProcessor, eventHubClientBuilder, false, tracerProvider); PartitionBasedLoadBalancer loadBalancer = new PartitionBasedLoadBalancer(checkpointStore, eventHubAsyncClient, fqNamespace, eventHubName, consumerGroupName, "owner", TimeUnit.SECONDS.toSeconds(5), - partitionPumpManager); + partitionPumpManager, ec -> { + }); loadBalancer.loadBalance(); sleep(5); verify(eventHubAsyncClient, atLeast(1)).getPartitionIds(); @@ -362,10 +363,11 @@ public void testEventHubClientFailure() { List partitionIds = new ArrayList<>(); when(eventHubAsyncClient.getPartitionIds()).thenReturn(Flux.fromIterable(partitionIds)); PartitionPumpManager partitionPumpManager = new PartitionPumpManager(checkpointStore, - () -> partitionProcessor, EventPosition.earliest(), eventHubClientBuilder, false, tracerProvider); + () -> partitionProcessor, eventHubClientBuilder, false, tracerProvider); PartitionBasedLoadBalancer loadBalancer = new PartitionBasedLoadBalancer(checkpointStore, eventHubAsyncClient, fqNamespace, eventHubName, consumerGroupName, "owner", TimeUnit.SECONDS.toSeconds(5), - partitionPumpManager); + partitionPumpManager, ec -> { + }); loadBalancer.loadBalance(); sleep(2); verify(eventHubAsyncClient, atLeast(1)).getPartitionIds(); @@ -458,8 +460,9 @@ public void processError(ErrorContext eventProcessingErrorContext) { eventProcessingErrorContext.getPartitionContext().getPartitionId(), eventProcessingErrorContext.getThrowable()); } - }, EventPosition.earliest(), eventHubClientBuilder, false, tracerProvider); + }, eventHubClientBuilder, false, tracerProvider); return new PartitionBasedLoadBalancer(checkpointStore, eventHubAsyncClient, fqNamespace, - eventHubName, consumerGroupName, owner, TimeUnit.SECONDS.toSeconds(5), partitionPumpManager); + eventHubName, consumerGroupName, owner, TimeUnit.SECONDS.toSeconds(5), partitionPumpManager, + ec -> { }); } }