From de5aaec98f7c5e03b8c3940b9a30366d710f99af Mon Sep 17 00:00:00 2001 From: Eric Honer Date: Tue, 13 Jun 2023 20:05:09 -0700 Subject: [PATCH] - Fix parameter ReceiverOptions parameters usaged - Change powermock parameters --- .../system/eventhub/consumer/EventHubSystemConsumer.java | 9 +++++---- .../eventhub/MockEventHubClientManagerFactory.java | 2 +- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java b/samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java index 91eab33eef..6fe2c72daf 100644 --- a/samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java +++ b/samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java @@ -283,9 +283,10 @@ private synchronized void initializeEventHubsManagers() { // If no such offset exists Eventhub will return an error. ReceiverOptions receiverOptions = new ReceiverOptions(); receiverOptions.setPrefetchCount(prefetchCount); + EventPosition position = EventPosition.fromOffset(offset, /* inclusiveFlag */false); receiver = eventHubClientManager.getEventHubClient() - .createReceiver(consumerGroup, partitionId.toString(), - EventPosition.fromOffset(offset, /* inclusiveFlag */false)).get(DEFAULT_EVENTHUB_CREATE_RECEIVER_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS); + .createReceiver(consumerGroup, partitionId.toString(), position, receiverOptions) + .get(DEFAULT_EVENTHUB_CREATE_RECEIVER_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS); } PartitionReceiveHandler handler = @@ -378,9 +379,9 @@ private void renewPartitionReceiver(SystemStreamPartition ssp) { // Recreate receiver ReceiverOptions receiverOptions = new ReceiverOptions(); receiverOptions.setPrefetchCount(prefetchCount); + EventPosition position = EventPosition.fromOffset(offset, !offset.equals(EventHubSystemConsumer.START_OF_STREAM)); PartitionReceiver receiver = eventHubClientManager.getEventHubClient() - .createReceiverSync(consumerGroup, partitionId.toString(), - EventPosition.fromOffset(offset, !offset.equals(EventHubSystemConsumer.START_OF_STREAM)), receiverOptions); + .createReceiverSync(consumerGroup, partitionId.toString(), position, receiverOptions); // Timeout for EventHubClient receive receiver.setReceiveTimeout(DEFAULT_EVENTHUB_RECEIVER_TIMEOUT); diff --git a/samza-azure/src/test/java/org/apache/samza/system/eventhub/MockEventHubClientManagerFactory.java b/samza-azure/src/test/java/org/apache/samza/system/eventhub/MockEventHubClientManagerFactory.java index 2d071518c2..240a1d5191 100644 --- a/samza-azure/src/test/java/org/apache/samza/system/eventhub/MockEventHubClientManagerFactory.java +++ b/samza-azure/src/test/java/org/apache/samza/system/eventhub/MockEventHubClientManagerFactory.java @@ -138,7 +138,7 @@ private class MockEventHubClientManager implements EventHubClientManager { return CompletableFuture.completedFuture(mockPartitionReceiver); }); - PowerMockito.when(mockEventHubClient.createReceiver(anyString(), anyString(), anyObject())) + PowerMockito.when(mockEventHubClient.createReceiver(anyString(), anyString(), anyObject(), anyObject())) .then((Answer>) invocationOnMock -> { String partitionId = invocationOnMock.getArgumentAt(1, String.class); EventPosition offset = invocationOnMock.getArgumentAt(2, EventPosition.class);