- Fix parameter ReceiverOptions parameters usaged (#1670)

- Change powermock parameters
apache · Jun 14, 2023 · 03d70a5 · 03d70a5
1 parent 9111bb7
commit 03d70a5
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 5 deletions.
diff --git a/...azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java b/...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/...zure/src/test/java/org/apache/samza/system/eventhub/MockEventHubClientManagerFactory.java b/...zure/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<CompletableFuture<PartitionReceiver>>) invocationOnMock -> {
                 String partitionId = invocationOnMock.getArgumentAt(1, String.class);
                 EventPosition offset = invocationOnMock.getArgumentAt(2, EventPosition.class);