SAMZA-1438: SystemProducer, Consumer and Admin interfaces for EventHubs #308

dxichen · 2017-09-29T22:36:53Z

No description provided.

vjagadish1989

Thanks for pulling this major refactor. This has tremendous leverage for Samza customers!

vjagadish1989 · 2017-10-03T22:11:14Z

...-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubEntityConnection.java

+  private final String sasKey;
+  private final String consumerName;
+  private final EventHubConfig config;
+  private final Map<Integer, PartitionReceiver> receivers = new TreeMap<>();


Currently, the EventHubEntityConnection class has some stateful/mutating operations -
including, some synchronized methods:

+ synchronized void addPartition(int partitionId, String offset, PartitionReceiveHandler handler) {} + synchronized void connectAndStart() {} + synchronized void stop() {}

Can we consider separating these and making the class immutable:

+ private final String namespace; + private final String entityPath; + private final String sasKeyName; + private final String sasKey; + private final String consumerName; + private final EventHubConfig config;

Also, I'd remove these state-ful fields from this class:

+ private final Map<Integer, PartitionReceiver> receivers = new TreeMap<>(); + private EventHubClientWrapper ehClientWrapper; + private final EventHubClientFactory eventHubClientFactory = new EventHubClientFactory();

Do you think we can move this map of partition -> PartitionReceiver instances into the Consumer itself since all of these are related to receiving events? Another nice thing is that you can get rid off the EventHubEntitityConnectionFactory completely for wiring these stateful components

Refactored in order to remove EventHubConnectionEnitity 1c1677f

vjagadish1989 · 2017-10-03T22:21:18Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubSystemFactory.java

+
+  @Override
+  public SystemProducer getProducer(String systemName, Config config, MetricsRegistry registry) {
+    return new EventHubSystemProducer(systemName, new EventHubConfig(config, systemName), registry);


Can we inject the EventHubWrapperFactory into the EventHubSystemProducer, EventHubSystemConsumer and EventHubSystemAdmin
instead of class-loading them via config? It'd be much cleaner with types that way. Use-cases at LinkedIn that require tunneling can inject their tunnel-aware versions of the EventHubClientWrapper.

vjagadish1989 · 2017-10-03T22:22:27Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java

+      if (events != null) {
+
+        events.forEach(event -> {
+            byte[] decryptedBody = event.getBytes();


Remove references to encryption and de-cryption everywhere? since they seem orthogonal to the record payloadss

Removed the Serde configs and added the Stream to Serde mapping to EventHubSystemFactory. (Passthrough Serde by default) e740088

vjagadish1989

I did another pass. will review test code soon! let's try wrapping this up by end of week.

vjagadish1989 · 2017-10-11T18:00:21Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventDataWrapper.java

+   * @return the body of decrypted body of the message. In case no encryption is setup for this topic,
+   * just returns the body of the message.
+   */
+  public byte[] getDecryptedBody() {


s/getDecryptedBody/getBody

vjagadish1989 · 2017-10-11T18:01:38Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubClientWrapper.java

+
+import com.microsoft.azure.eventhubs.EventHubClient;
+
+public interface EventHubClientWrapper {


Add java docs:
Wraps the {@link EventHubClient} with lifecycle hooks for initialization and close

vjagadish1989 · 2017-10-11T18:04:11Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubConfig.java

+  public static final String DEFAULT_CONFIG_STREAM_CONSUMER_START_POSITION = StartPosition.LATEST.name();
+
+  public static final String CONFIG_PRODUCER_PARTITION_METHOD = "systems.%s.eventhubs.partition.method";
+  public static final String DEFAULT_CONFIG_PRODUCER_PARTITION_METHOD = EventHubSystemProducer


Can't we internally translate calls to send() into one of these partition schemes? For instance, if a specific partitionId is provided, we write it to the partition. If not, we use the eventhub's hash partitioner?

the partitionId is obtained by the partitionKey on the outgoingEnvelope. This id could be used as the partition on eventhub or the seed for the consistent hashing on eventhub.

It could be ambiguous if the user wants to use event hub hash by providing the hash via the partitionId if the partition method is not specified.

vjagadish1989 · 2017-10-11T18:05:36Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubConfig.java

+  public static final String CONFIG_SEND_KEY_IN_EVENT_PROPERTIES = "systems.%s.eventhubs.send.key";
+  public static final Boolean DEFAULT_CONFIG_SEND_KEY_IN_EVENT_PROPERTIES = false;
+
+  public static final String CONFIG_CONNECTION_SHUTDOWN_TIMEOUT_MILLIS = "systems.%s.eventhubs.shutdown.timeout";


Would be good to confirm if all these configs need to be exposed as top-level configs? If we can simply propagate eventhub configs down to the client, it'll save us from maintaining these configs

vjagadish1989 · 2017-10-11T18:07:06Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubConfig.java

+
+  private final String systemName;
+
+  public EventHubConfig(Map<String, String> config, String systemName) {


Is there a reason for EventHubConfig to take in a systemName? Seems fairly system agnostic.

vjagadish1989 · 2017-10-11T18:15:31Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/admin/EventHubSystemAdmin.java

+                    TimeUnit.MILLISECONDS);
+            streamPartitions.put(streamName, ehInfo.getPartitionIds());
+          } catch (InterruptedException | ExecutionException e) {
+            String msg = String.format("Error while fetching EventHubRuntimeInfo for System:%s, Stream:%s",


Cant we use the same path for all exceptions? We appear to be handling exceptions the same way in either of these catch code paths.

vjagadish1989 · 2017-10-11T18:16:32Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/admin/EventHubSystemAdmin.java

+            throw new SamzaException(msg);
+          }
+        }
+        requestedMetadata.put(streamName, new SystemStreamMetadata(streamName,


Where is close called? If not, we should close the client in finally block to avoid memory leaks?

vjagadish1989 · 2017-10-11T18:17:17Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java

+  public static final String READ_LATENCY = "readLatency";
+  public static final String READ_ERRORS = "readErrors";
+
+  private static Counter aggEventReadRate = null;


Can these counters be per-instance variables? or do they have to be shared? another nice thing is that you can get rid off this aggregate lock.

cc: @lhaiesp to confirm

vjagadish1989 · 2017-10-11T18:17:48Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java

+  private static SamzaHistogram aggReadLatency = null;
+  private static Counter aggReadErrors = null;
+
+  private Map<String, Counter> eventReadRates;


consider making these maps final

vjagadish1989 · 2017-10-11T18:19:12Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java

+    readErrors = streamList.stream()
+            .collect(Collectors.toMap(Function.identity(), x -> registry.newCounter(x, READ_ERRORS)));
+
+    // Locking to ensure that these aggregated metrics will be created only once across multiple system consumers.


Do we really need an aggregate view of the metric?

@lhaiesp

vjagadish1989 · 2017-10-11T18:28:38Z

@prateekm.

cc:
@lhaiesp @srinipunuru for review as well since they are knowledgeable about Eventhubs.

prateekm

This PR has somewhat different formatting (esp. continuation indent and param alignment) than Samza conventions. Would prefer making it consistent for readability.

prateekm · 2017-10-11T21:52:38Z

gradle/wrapper/gradle-wrapper.properties

 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-2.8-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-2.8-all.zip


Why is this required?

Nope, not sure why it's there

This actually helps you get all the gradle dependencies. So you that you can browse through the gradle plugins from your IDE.

prateekm · 2017-10-11T21:53:01Z

build.gradle

  }
  checkstyle {
    configFile = new File(rootDir, "checkstyle/checkstyle.xml")
    toolVersion = "$checkstyleVersion"
  }
+  test {
+    exclude 'org/apache/samza/system/eventhub/producer/**'


Why is this required?

Excluding it since it is an integration test that requires access to an external EventHub instance

prateekm · 2017-10-11T21:53:40Z

build.gradle

    compile project(':samza-api')
    compile project(":samza-core_$scalaVersion")
    compile "org.slf4j:slf4j-api:$slf4jVersion"
    testCompile "junit:junit:$junitVersion"
+    testCompile "org.mockito:mockito-all:$mockitoVersion"
+    testCompile "org.powermock:powermock-api-mockito:$powerMockVersion"


Prefer not using powermock unless we really need it.

Unfortunately, I need powermock in order to mock the EventHubClient (private ctor) and PartitionReceiver(final class) for unit tests.

prateekm · 2017-10-11T21:54:26Z

gradle/wrapper/gradle-wrapper.properties

@@ -1,6 +1,6 @@
-#Mon Oct 31 23:13:44 PDT 2016
+#Thu Sep 07 09:55:54 PDT 2017


Can remove this, not sure why this is here.

prateekm · 2017-10-11T21:57:52Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventDataWrapper.java

+/**
+ * Simpler wrapper of {@link EventData} events with the decrypted payload
+ */
+public class EventDataWrapper {


Why is this wrapper class necessary? Is this just for holding the decrypted body? If so, maybe DecryptedEventData extends EventData or just use an IME?

EventData cannot be extended due to private ctor. This is used as a workaround to pass to both the metadata from EventData and the decrypted body as a message to IME.

prateekm · 2017-10-12T00:04:55Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/producer/EventHubSystemProducer.java

+            ehClient.getEventHubClient());
+
+    Instant endTime = Instant.now();
+    long latencyMs = Duration.between(startTime, endTime).toMillis();


Maybe just use currentTimeMillis/nanoTime and subtraction for simplicity?

prateekm · 2017-10-12T00:07:15Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/producer/EventHubSystemProducer.java

+    if (partitioningMethod == PartitioningMethod.EVENT_HUB_HASHING) {
+      return eventHubClient.send(eventData, convertPartitionKeyToString(partitionKey));
+    } else if (partitioningMethod == PartitioningMethod.PARTITION_KEY_AS_PARTITION) {
+      if (!(partitionKey instanceof Integer)) {


Null check too (or is it guaranteed to be not-null?).

null check is not necessary for instanceof

prateekm · 2017-10-12T00:08:12Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/producer/EventHubSystemProducer.java

+      }
+
+      PartitionSender sender = getPartitionSender(streamName, (int) partitionKey, eventHubClient);
+      return sender.send(eventData);


Is there a built in back-pressure mechanism (producer buffer size configuration etc.) somewhere? Would we run OOM if the producer can't keep up?

prateekm · 2017-10-12T00:08:45Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/producer/EventHubSystemProducer.java

+    }
+  }
+
+  private Object getEnvelopePartitionId(OutgoingMessageEnvelope envelope) {


Unnecessary, can inline.

Internal implementation will override this, will make protected

prateekm · 2017-10-12T00:09:12Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/producer/EventHubSystemProducer.java

+    }
+  }
+
+  private PartitionSender getPartitionSender(String streamName, int partition, EventHubClient eventHubClient) {


getOrCreate

Why is the creation happening during send? Should it happen once during start instead?

The partition key is contained in the OME. This is required to determine the partition of the partitionSender created. Will create for all stream partition downstream on create.

prateekm · 2017-10-13T18:27:52Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/producer/EventHubSystemProducer.java

+          sendErrors.get(destination).inc();
+          aggSendErrors.inc();
+          LOG.error("Send message to event hub: {} failed with exception: ", destination, throwable);
+          sendExceptionOnCallback = throwable;


If this handler isn't called on the send thread (I assume it's called on the thread that completed the future?), this exception may still not be visible to the sender threads even if the send method is synchronized:

https://www.cs.umd.edu/~pugh/java/memoryModel/jsr-133-faq.html#synchronization
"Synchronization ensures that memory writes by a thread before or during a synchronized block are made visible in a predictable manner to other threads which synchronize on the same monitor."

Good catch, I will to sync on the sendExceptionOnCallback throwable instead.

shanthoosh

Thanks for the patch.
Just few questions to understand the patch(and minor suggestions).

shanthoosh · 2017-10-19T15:18:13Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/admin/EventHubSystemAdmin.java

+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class EventHubSystemAdmin implements SystemAdmin {


@vjagadish1989
Just curious.
Do we need this to implement ExtendedSystemAdmin here. It's usages revolves around in TaskStorageManager and StremaMetadataCache. Is it not required for EventHub scenario?

Seems like there will be no problem in implementing the extended version, but there might be something I'm missing. Maybe we can discuss this offline.

shanthoosh · 2017-10-19T15:19:44Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/SamzaEventHubClient.java

+/**
+ * Wraps the {@link EventHubClient} with lifestyle hooks for initialization and close.
+ */
+public interface SamzaEventHubClient {


Just wondering if it's logical to rename this to EventHubManager or EventHubClientManager.

SamzaEventHubClient.getEventHubClient looks odd.

I agree it looks a bit odd, EventHubClientManager works!

shanthoosh · 2017-10-19T15:28:20Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java

+            .collect(Collectors.toMap(Function.identity(), x -> registry.newCounter(x, READ_ERRORS)));
+
+    // Locking to ensure that these aggregated metrics will be created only once across multiple system consumers.
+    synchronized (AGGREGATE) {


It's not recommended to lock on Strings. If the samza user class uses some synchronization based upon same string literals there could be potential cases of deadlock.

It's well documented here: https://docs.oracle.com/javase/specs/jls/se7/html/jls-3.html#jls-3.10.5

Literal strings within different classes in different packages represent references to the same String object.

It would be safe to create a static object and lock here based upon that (and use Double checked locking or make the eventReadRate as volatile). Or just remove this lock in itself and make metrics per instance.

Nice catch, will use Object instead

shanthoosh · 2017-10-19T15:38:57Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java

+  private final ConcurrentHashMap<String, SamzaEventHubClient> streamEventHubClients = new ConcurrentHashMap<>();
+  private final ConcurrentHashMap<SystemStreamPartition, String> streamPartitionOffsets = new ConcurrentHashMap<>();
+  private final Map<String, Serde<byte[]>> serdes;
+  private boolean isStarted = false;


Does this have to be volatile?

shanthoosh · 2017-10-19T15:45:38Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/metrics/SamzaHistogram.java

+ * Creates a {@link Histogram} metric using {@link ExponentiallyDecayingReservoir}
+ * Keeps a {@link Gauge} for each percentile
+ */
+public class SamzaHistogram {


Please correct me if i'm wrong here.
All samza metric classes(Timer, Counter etc) is a subtype of Metric. Doesn't this have to follow the same class hierarchy. I think some Metricreporters would not be abel to report this metric otherwise.

As mentioned above, I was planning to have another PR for the metrics histogram update, thanks for the comment!

@shanthoosh You are right. this PR is intentionally kept separate from that.

shanthoosh · 2017-10-19T15:49:04Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java

+      if (aggEventReadRate == null) {
+        aggEventReadRate = registry.newCounter(AGGREGATE, EVENT_READ_RATE);
+        aggEventByteReadRate = registry.newCounter(AGGREGATE, EVENT_BYTE_READ_RATE);
+        aggReadLatency = new SamzaHistogram(registry, AGGREGATE, READ_LATENCY);


Please correct me if i'm wrong here.

Instead of using new Samzahistogram here, don't we have to do something like registry.newHistogram(...).
I think we need to introduce newHistogram method in MetricsRegistry interface for using this new metric and reporters to function.

I was planning to add the histogram as is for now and have a new PR later on to properly add the histogram metric.

shanthoosh · 2017-10-19T15:54:33Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/producer/EventHubSystemProducer.java

+    }
+
+    // Locking to ensure that these aggregated metrics will be created only once across multiple system consumers.
+    synchronized (AGGREGATE) {


If i understand right, the goal is to achieve one time aggregate metrics initialization across all instances of EventhubProducer. synchronized in method name just guarantees lock per EventhubProducer instance.

shanthoosh · 2017-10-19T16:09:02Z

build.gradle

  }
  checkstyle {
    configFile = new File(rootDir, "checkstyle/checkstyle.xml")
    toolVersion = "$checkstyleVersion"
  }
+  test {


Just curious. Any reason to exclude these tests from the build?

They are integration tests that require a connection to an external EventHub. Is there a Samza-specific protocol for integration test? Will add a comment here though

@shanthoosh these tests connect to eventhub, and we'll need to expose credentials to do that. So, we've intentionally disabled them

shanthoosh · 2017-10-19T16:38:47Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java

+                  config.getStartPosition(systemName, systemStreamPartition.getStream()));
+      }
+    }
+    streamPartitionOffsets.put(systemStreamPartition, offset);


Please correct me if i'm wrong here.

I think this has to use offsetComparator and pick the lowest offset.

If we've two tasks with checkpointed offsets T1(SP1: START_OF_STREAM), T2(SP1: END_OF_STREAM), register should choose the lowest one and store in streamPartitionOffsets for a partition(This register call will be called once for each TaskInstance). If we just overwrite, we could pick End (depending upon register call order).

Good catch, I was relying on the register being only called once per SSP

vjagadish1989

looks great. approved.

thank you for the contribution.

vjagadish1989 · 2017-10-25T05:35:48Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubClientManager.java

+import com.microsoft.azure.eventhubs.EventHubClient;
+
+/**
+ * Wraps the {@link EventHubClient} with lifestyle hooks for initialization and close.


s/lifestyle/lifecycle

{@link #init} should be invoked once during the startup and provides a hook to perform some initialization before the underlying {@link EventHubClient} client is created. {@link #close} is invoked once during shut-down and can be used to perform clean-ups.

vjagadish1989 · 2017-10-25T05:46:10Z

build.gradle

  }
  checkstyle {
    configFile = new File(rootDir, "checkstyle/checkstyle.xml")
    toolVersion = "$checkstyleVersion"
  }
+  test {


@shanthoosh these tests connect to eventhub, and we'll need to expose credentials to do that. So, we've intentionally disabled them

vjagadish1989 · 2017-10-25T05:56:06Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubClientManager.java

+ */
+public interface EventHubClientManager {
+  /**
+   * Initiate the connection to EventHub.


Reword docs
Lifecycle hook to perform initializations before the {@link EventHubClient} is created.

Since this also creates the EventHubClient, how about
Lifecycle hook to perform initializations for the creation of the underlying {@link EventHubClient}.

vjagadish1989 · 2017-10-25T05:58:03Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubClientManager.java

+   */
+  void init();
+
+  /**


Reword docs:
Returns the underlying {@link EventHubClient} instance. Multiple invocations of {@link #getEventHubClient} should return the same instance instead of creating new ones.

vjagadish1989 · 2017-10-25T05:59:43Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubClientManager.java

+  EventHubClient getEventHubClient();
+
+  /**
+   * Timed synchronous connection close to the EventHub.


Reword docs
Closes the {@link EventHubClient} instance. Use {@link#close} to perform clean-ups after the execution of the {@link EventHubClient}.

vjagadish1989 · 2017-10-25T06:25:48Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java

+    return super.poll(systemStreamPartitions, timeout);
+  }
+
+  private void renewPartitionReceiver(SystemStreamPartition ssp, Throwable throwable) {


Can log the exception at the caller, and not take the throwable here.

vjagadish1989 · 2017-10-25T06:26:28Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java

+      // Timeout for EventHubClient receive
+      receiver.setReceiveTimeout(DEFAULT_EVENTHUB_RECEIVER_TIMEOUT);
+      receiver.setReceiveHandler(streamPartitionHandlers.get(ssp));
+      streamPartitionReceivers.put(ssp, receiver);


Should the receiver be started? Or is it done automatically>

This is started by setReceiveHandler, will add comment

vjagadish1989 · 2017-10-25T06:29:27Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java

+    public void onReceive(Iterable<EventData> events) {
+      if (events != null) {
+        events.forEach(event -> {
+            byte[] decryptedBody = event.getBytes();


Remove all references to encryption/decryption. Should define a message interceptor instead.

public interface Interceptor { byte[] intercept(byte[] input) }

We will define interceptors for encryption and decryption at LinkedIn optionally. We'll have a pass-through interceptor for the open source. Another nice thing is that, Serde continues to happen inside Samza.

vjagadish1989 · 2017-10-25T06:32:10Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java

+
+  @Override
+  public Map<SystemStreamPartition, List<IncomingMessageEnvelope>> poll(Set<SystemStreamPartition> systemStreamPartitions, long timeout) throws InterruptedException {
+    if (perminentEventHubError != null) {


s/perminent/permenant

you mean permanent? :)

vjagadish1989 · 2017-10-25T06:33:39Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/metrics/SamzaHistogram.java

+ * Creates a {@link Histogram} metric using {@link ExponentiallyDecayingReservoir}
+ * Keeps a {@link Gauge} for each percentile
+ */
+public class SamzaHistogram {


@shanthoosh You are right. this PR is intentionally kept separate from that.

vjagadish1989 reviewed Oct 3, 2017

View reviewed changes

vjagadish1989 reviewed Oct 11, 2017

View reviewed changes

prateekm requested changes Oct 12, 2017

View reviewed changes

prateekm reviewed Oct 13, 2017

View reviewed changes

shanthoosh reviewed Oct 19, 2017

View reviewed changes

vjagadish1989 approved these changes Oct 25, 2017

View reviewed changes

dxichen added 23 commits October 26, 2017 17:53

added gradle changes

507e91d

initial commit, consumer, producer, admin

120e182

fixed indents, reformated some code

cc6eda0

added admin tests, factored config creation

39a5392

factored producer configs

9a9012c

cleanup producer, added test

c1e5bfd

abstract eventhub client wrapper

3aad90b

fix code style, added timeout configs

f2d3aa3

removed depricated EventData methods

2c22671

throw on missing connection configs, fixed typos

d0b1f76

removed unused import

196963b

more accurate fetch for newestOffset

4343678

leverage EH constant partition size to decrease request

0f9065d

added Event Hub consumer buffer capacity per SystemStreamPartition

1ebc467

Removed EHEntityConnection, Moved EHWrapper factory to System ctors

aab7f51

ignore intergration test

bfb48b7

compatibility with upstream changes

097f355

moved per stream Serde setting to EHSystemFactory

41016b7

extract method of partition key determination

61e3dbb

Consistent timeout results

1039928

Simple pr comments addressed

e3e21f6

Moved create logic for Producer to start, added unit tests for Producer

f514ecd

Removed EventDataWrapper, added EventHubIME

2b2c11c

dxichen added 11 commits October 26, 2017 17:53

Added admin close, consumer async onerror callback, producer sync

0558660

Renaming SamzaEventHubClient, added Consumer retry on transient error

c7e69dd

Support multiple register consumer calls to same stream

e958073

Support multiple register producer calls to same stream

a452a8e

Save first throwable

6c0aa47

Removed starting offset default in EH consumer, use system level config

ac62987

minor pr comments and docs

503df57

Log warning on poll when a non transient error occured

811b90f

Added interceptor, renamed EventHubIME

9a4290a

Close PartitionReceiver before creating a new one

705f0f3

Added extra comment for receicer start clarity

ea1bafe

dxichen force-pushed the eventHub-Connector branch from 7bc10d2 to ea1bafe Compare October 27, 2017 00:57

asfgit closed this in 663b8da Oct 27, 2017


		import com.microsoft.azure.eventhubs.EventHubClient;

		public interface EventHubClientWrapper {


		private final String systemName;

		public EventHubConfig(Map<String, String> config, String systemName) {

		@@ -1,6 +1,6 @@
		#Mon Oct 31 23:13:44 PDT 2016
		#Thu Sep 07 09:55:54 PDT 2017

SAMZA-1438: SystemProducer, Consumer and Admin interfaces for EventHubs #308

SAMZA-1438: SystemProducer, Consumer and Admin interfaces for EventHubs #308

Conversation

dxichen commented Sep 29, 2017

vjagadish1989 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dxichen Oct 9, 2017 • edited Loading

Choose a reason for hiding this comment

vjagadish1989 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vjagadish1989 commented Oct 11, 2017 • edited Loading

prateekm left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dxichen Oct 12, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shanthoosh left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dxichen Oct 19, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vjagadish1989 left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vjagadish1989 Oct 25, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vjagadish1989 Oct 25, 2017 • edited Loading

dxichen Oct 9, 2017 •

edited

Loading

vjagadish1989 commented Oct 11, 2017 •

edited

Loading

dxichen Oct 12, 2017 •

edited

Loading

dxichen Oct 19, 2017 •

edited

Loading

vjagadish1989 left a comment •

edited

Loading

vjagadish1989 Oct 25, 2017 •

edited

Loading

vjagadish1989 Oct 25, 2017 •

edited

Loading

vjagadish1989 Oct 25, 2017 •

edited

Loading