You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by ja...@apache.org on 2018/04/24 19:26:18 UTC
samza git commit: SAMZA-1676: miscellaneous fix and improvement for
eventhubs system
Repository: samza
Updated Branches:
refs/heads/master 7a32d30db -> e63d27c0c
SAMZA-1676: miscellaneous fix and improvement for eventhubs system
Including these changes:
- Log the metadata that we are fetching from the event hubs
- Rename readLatency to consumptionLagMs
- fix the issue that readLatency metric returns negative value
Author: Hai Lu <ha...@linkedin.com>
Reviewers: Srini P<sp...@linkedin.com>, Jagadish <ja...@apache.org>
Closes #484 from lhaiesp/master
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/e63d27c0
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/e63d27c0
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/e63d27c0
Branch: refs/heads/master
Commit: e63d27c0cd08973f35ae5c6967f71ddd3dff31c3
Parents: 7a32d30
Author: Hai Lu <ha...@linkedin.com>
Authored: Tue Apr 24 12:26:14 2018 -0700
Committer: Jagadish <jv...@linkedin.com>
Committed: Tue Apr 24 12:26:14 2018 -0700
----------------------------------------------------------------------
.../eventhub/SamzaEventHubClientManager.java | 2 +
.../eventhub/admin/EventHubSystemAdmin.java | 47 ++++++++++++++++++--
.../consumer/EventHubSystemConsumer.java | 28 +++++++-----
3 files changed, 63 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/e63d27c0/samza-azure/src/main/java/org/apache/samza/system/eventhub/SamzaEventHubClientManager.java
----------------------------------------------------------------------
diff --git a/samza-azure/src/main/java/org/apache/samza/system/eventhub/SamzaEventHubClientManager.java b/samza-azure/src/main/java/org/apache/samza/system/eventhub/SamzaEventHubClientManager.java
index a884a79..22621f3 100644
--- a/samza-azure/src/main/java/org/apache/samza/system/eventhub/SamzaEventHubClientManager.java
+++ b/samza-azure/src/main/java/org/apache/samza/system/eventhub/SamzaEventHubClientManager.java
@@ -76,6 +76,7 @@ public class SamzaEventHubClientManager implements EventHubClientManager {
@Override
public void init() {
String remoteHost = String.format(EVENTHUB_REMOTE_HOST_FORMAT, eventHubNamespace);
+ LOG.info("Initializing SamzaEventHubClientManager for namespace: " + eventHubNamespace);
try {
ConnectionStringBuilder connectionStringBuilder = new ConnectionStringBuilder()
.setNamespaceName(eventHubNamespace)
@@ -92,6 +93,7 @@ public class SamzaEventHubClientManager implements EventHubClientManager {
LOG.error(msg, e);
throw new SamzaException(msg, e);
}
+ LOG.info("SamzaEventHubClientManager initialized for namespace: " + eventHubNamespace);
}
@Override
http://git-wip-us.apache.org/repos/asf/samza/blob/e63d27c0/samza-azure/src/main/java/org/apache/samza/system/eventhub/admin/EventHubSystemAdmin.java
----------------------------------------------------------------------
diff --git a/samza-azure/src/main/java/org/apache/samza/system/eventhub/admin/EventHubSystemAdmin.java b/samza-azure/src/main/java/org/apache/samza/system/eventhub/admin/EventHubSystemAdmin.java
index 344a6a8..c86e31a 100644
--- a/samza-azure/src/main/java/org/apache/samza/system/eventhub/admin/EventHubSystemAdmin.java
+++ b/samza-azure/src/main/java/org/apache/samza/system/eventhub/admin/EventHubSystemAdmin.java
@@ -19,6 +19,7 @@
package org.apache.samza.system.eventhub.admin;
+import com.microsoft.azure.eventhubs.EventHubClient;
import com.microsoft.azure.eventhubs.EventHubRuntimeInformation;
import com.microsoft.azure.eventhubs.PartitionRuntimeInformation;
import java.util.Arrays;
@@ -76,6 +77,33 @@ public class EventHubSystemAdmin implements SystemAdmin {
return results;
}
+ // EventHubRuntimeInformation does not implement toString()
+ private String printEventHubRuntimeInfo(EventHubRuntimeInformation ehInfo) {
+ if (ehInfo == null) {
+ return "[EventHubRuntimeInformation: null]";
+ }
+ return String.format("[EventHubRuntimeInformation: createAt=%s, partitionCount=%d, path=%s]", ehInfo.getCreatedAt(),
+ ehInfo.getPartitionCount(), ehInfo.getPath());
+ }
+
+ // PartitionRuntimeInformation does not implement toString()
+ private String printPartitionRuntimeInfo(PartitionRuntimeInformation runtimeInformation) {
+ if (runtimeInformation == null) {
+ return "[PartitionRuntimeInformation: null]";
+ }
+ StringBuilder stringBuilder = new StringBuilder();
+ stringBuilder.append("[PartitionRuntimeInformation:");
+ stringBuilder.append(" eventHubPath=").append(runtimeInformation.getEventHubPath());
+ stringBuilder.append(" partitionId=").append(runtimeInformation.getPartitionId());
+ stringBuilder.append(" lastEnqueuedTimeUtc=").append(runtimeInformation.getLastEnqueuedTimeUtc().toString());
+ stringBuilder.append(" lastEnqueuedOffset=").append(runtimeInformation.getLastEnqueuedOffset());
+ // calculate the number of messages in the queue
+ stringBuilder.append(" numMessages=")
+ .append(runtimeInformation.getLastEnqueuedSequenceNumber() - runtimeInformation.getBeginSequenceNumber());
+ stringBuilder.append("]");
+ return stringBuilder.toString();
+ }
+
@Override
public Map<String, SystemStreamMetadata> getSystemStreamMetadata(Set<String> streamNames) {
Map<String, SystemStreamMetadata> requestedMetadata = new HashMap<>();
@@ -86,11 +114,24 @@ public class EventHubSystemAdmin implements SystemAdmin {
LOG.debug(String.format("Partition ids for Stream=%s not found", streamName));
EventHubClientManager eventHubClientManager = getOrCreateStreamEventHubClient(streamName);
- CompletableFuture<EventHubRuntimeInformation> runtimeInfo = eventHubClientManager.getEventHubClient().getRuntimeInformation();
+ EventHubClient ehClient = eventHubClientManager.getEventHubClient();
+ CompletableFuture<EventHubRuntimeInformation> runtimeInfo = ehClient.getRuntimeInformation();
long timeoutMs = eventHubConfig.getRuntimeInfoWaitTimeMS(systemName);
EventHubRuntimeInformation ehInfo = runtimeInfo.get(timeoutMs, TimeUnit.MILLISECONDS);
- LOG.info(String.format("Adding partition ids=%s for stream=%s", Arrays.toString(ehInfo.getPartitionIds()), streamName));
+ LOG.info(String.format("Adding partition ids=%s for stream=%s. EHRuntimetInfo=%s",
+ Arrays.toString(ehInfo.getPartitionIds()), streamName, printEventHubRuntimeInfo(ehInfo)));
+
+ try {
+ for (String partitionId : ehInfo.getPartitionIds()) {
+ LOG.info(printPartitionRuntimeInfo(
+ ehClient.getPartitionRuntimeInformation(partitionId).get(timeoutMs, TimeUnit.MILLISECONDS)));
+ }
+ } catch (Exception e) {
+ // ignore failures as this is just for information logging
+ LOG.warn("Failed to fetch and print partition runtime info from EventHubs.", e);
+ }
+
streamPartitions.put(streamName, ehInfo.getPartitionIds());
}
String[] partitionIds = streamPartitions.get(streamName);
@@ -113,7 +154,7 @@ public class EventHubSystemAdmin implements SystemAdmin {
private EventHubClientManager getOrCreateStreamEventHubClient(String streamName) {
if (!eventHubClients.containsKey(streamName)) {
- LOG.debug(String.format("Creating EventHubClient for Stream=%s", streamName));
+ LOG.info(String.format("Creating EventHubClient for Stream=%s", streamName));
EventHubClientManager eventHubClientManager = eventHubClientManagerFactory
.getEventHubClientManager(systemName, streamName, eventHubConfig);
http://git-wip-us.apache.org/repos/asf/samza/blob/e63d27c0/samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java
----------------------------------------------------------------------
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 f00944b..0fd0f2c 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
@@ -106,7 +106,7 @@ public class EventHubSystemConsumer extends BlockingEnvelopeMap {
public static final String END_OF_STREAM = "-2";
public static final String EVENT_READ_RATE = "eventReadRate";
public static final String EVENT_BYTE_READ_RATE = "eventByteReadRate";
- public static final String READ_LATENCY = "readLatency";
+ public static final String CONSUMPTION_LAG_MS = "consumptionLagMs";
public static final String READ_ERRORS = "readErrors";
public static final String AGGREGATE = "aggregate";
@@ -114,12 +114,12 @@ public class EventHubSystemConsumer extends BlockingEnvelopeMap {
private static Counter aggEventReadRate = null;
private static Counter aggEventByteReadRate = null;
- private static SamzaHistogram aggReadLatency = null;
+ private static SamzaHistogram aggConsumptionLagMs = null;
private static Counter aggReadErrors = null;
private final Map<String, Counter> eventReadRates;
private final Map<String, Counter> eventByteReadRates;
- private final Map<String, SamzaHistogram> readLatencies;
+ private final Map<String, SamzaHistogram> consumptionLagMs;
private final Map<String, Counter> readErrors;
final ConcurrentHashMap<SystemStreamPartition, PartitionReceiveHandler> streamPartitionHandlers =
@@ -162,8 +162,8 @@ public class EventHubSystemConsumer extends BlockingEnvelopeMap {
streamIds.stream().collect(Collectors.toMap(Function.identity(), x -> registry.newCounter(x, EVENT_READ_RATE)));
eventByteReadRates = streamIds.stream()
.collect(Collectors.toMap(Function.identity(), x -> registry.newCounter(x, EVENT_BYTE_READ_RATE)));
- readLatencies = streamIds.stream()
- .collect(Collectors.toMap(Function.identity(), x -> new SamzaHistogram(registry, x, READ_LATENCY)));
+ consumptionLagMs = streamIds.stream()
+ .collect(Collectors.toMap(Function.identity(), x -> new SamzaHistogram(registry, x, CONSUMPTION_LAG_MS)));
readErrors =
streamIds.stream().collect(Collectors.toMap(Function.identity(), x -> registry.newCounter(x, READ_ERRORS)));
@@ -172,7 +172,7 @@ public class EventHubSystemConsumer extends BlockingEnvelopeMap {
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);
+ aggConsumptionLagMs = new SamzaHistogram(registry, AGGREGATE, CONSUMPTION_LAG_MS);
aggReadErrors = registry.newCounter(AGGREGATE, READ_ERRORS);
}
}
@@ -222,6 +222,7 @@ public class EventHubSystemConsumer extends BlockingEnvelopeMap {
@Override
public void start() {
isStarted = true;
+ LOG.info("Starting EventHubSystemConsumer. Count of SSPs registered: " + streamPartitionOffsets.entrySet().size());
// Create receivers for Event Hubs
for (Map.Entry<SystemStreamPartition, String> entry : streamPartitionOffsets.entrySet()) {
@@ -258,7 +259,7 @@ public class EventHubSystemConsumer extends BlockingEnvelopeMap {
PartitionReceiveHandler handler =
new PartitionReceiverHandlerImpl(ssp, eventReadRates.get(streamId), eventByteReadRates.get(streamId),
- readLatencies.get(streamId), readErrors.get(streamId), interceptors.getOrDefault(streamId, null),
+ consumptionLagMs.get(streamId), readErrors.get(streamId), interceptors.getOrDefault(streamId, null),
config.getMaxEventCountPerPoll(systemName));
// Timeout for EventHubClient receive
@@ -274,8 +275,9 @@ public class EventHubSystemConsumer extends BlockingEnvelopeMap {
String.format("Failed to create receiver for EventHubs: namespace=%s, entity=%s, partitionId=%d", namespace,
entityPath, partitionId), e);
}
- LOG.debug(String.format("Connection successfully started for namespace=%s, entity=%s ", namespace, entityPath));
+ LOG.info(String.format("Connection successfully started for namespace=%s, entity=%s ", namespace, entityPath));
}
+ LOG.info("EventHubSystemConsumer started");
}
@Override
@@ -334,7 +336,7 @@ public class EventHubSystemConsumer extends BlockingEnvelopeMap {
@Override
public void stop() {
- LOG.debug("Stopping event hub system consumer...");
+ LOG.info("Stopping event hub system consumer...");
List<CompletableFuture<Void>> futures = new ArrayList<>();
streamPartitionReceivers.values().forEach((receiver) -> futures.add(receiver.close()));
CompletableFuture<Void> future = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
@@ -422,9 +424,9 @@ public class EventHubSystemConsumer extends BlockingEnvelopeMap {
eventByteReadRate.inc(eventDataLength);
aggEventByteReadRate.inc(eventDataLength);
- long latencyMs = Duration.between(Instant.now(), event.getSystemProperties().getEnqueuedTime()).toMillis();
+ long latencyMs = Duration.between(event.getSystemProperties().getEnqueuedTime(), Instant.now()).toMillis();
readLatency.update(latencyMs);
- aggReadLatency.update(latencyMs);
+ aggConsumptionLagMs.update(latencyMs);
}
@Override
@@ -440,12 +442,16 @@ public class EventHubSystemConsumer extends BlockingEnvelopeMap {
// Only set to transient throwable if there has been no previous errors
eventHubHandlerError.compareAndSet(null, throwable);
+ LOG.warn(
+ String.format("Received transient exception from EH client. Renew partition receiver for ssp: %s", ssp),
+ throwable);
// Retry creating a receiver since error likely due to timeout
renewPartitionReceiver(ssp);
return;
}
}
+ LOG.error(String.format("Received non transient exception from EH client for ssp: %s", ssp), throwable);
// Propagate non transient or unknown errors
eventHubHandlerError.set(throwable);
}