You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by ni...@apache.org on 2017/08/09 17:50:49 UTC
[03/23] samza git commit: SAMZA-1304: Handling duplicate stream
processor registration.
SAMZA-1304: Handling duplicate stream processor registration.
When a stream processor registers with same processorId as already existing
processor in processor group, it's registration should fail.
Author: Shanthoosh Venkataraman <sv...@linkedin.com>
Reviewers: Navina Ramesh <na...@apache.org>, Jagadish V <jv...@linkedin.com>
Closes #240 from shanthoosh/standalone_duplicate_processor_fix
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/45931fd5
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/45931fd5
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/45931fd5
Branch: refs/heads/0.14.0
Commit: 45931fd5d4406f731013b39d10fbfd577d6ac6a4
Parents: ebb1b7f
Author: Shanthoosh Venkataraman <sv...@linkedin.com>
Authored: Thu Jul 13 16:19:56 2017 -0700
Committer: navina <na...@apache.org>
Committed: Thu Jul 13 16:19:56 2017 -0700
----------------------------------------------------------------------
.../java/org/apache/samza/zk/ProcessorData.java | 18 ++-
.../org/apache/samza/zk/ZkJobCoordinator.java | 10 +-
.../main/java/org/apache/samza/zk/ZkUtils.java | 117 ++++++++++++++++---
.../java/org/apache/samza/zk/TestZkUtils.java | 36 +++++-
.../processor/TestZkLocalApplicationRunner.java | 13 ++-
5 files changed, 169 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/45931fd5/samza-core/src/main/java/org/apache/samza/zk/ProcessorData.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ProcessorData.java b/samza-core/src/main/java/org/apache/samza/zk/ProcessorData.java
index 3f4fd0b..a48a450 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ProcessorData.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ProcessorData.java
@@ -19,9 +19,12 @@
package org.apache.samza.zk;
+import java.util.Objects;
import org.apache.samza.SamzaException;
-
+/**
+ * Represents processor data stored in zookeeper processors node.
+ */
public class ProcessorData {
private final String processorId;
private final String host;
@@ -51,4 +54,17 @@ public class ProcessorData {
public String getProcessorId() {
return processorId;
}
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(processorId, host);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == null) return false;
+ if (getClass() != obj.getClass()) return false;
+ final ProcessorData other = (ProcessorData) obj;
+ return Objects.equals(processorId, other.processorId) && Objects.equals(host, other.host);
+ }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/45931fd5/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java b/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java
index 94c3054..8ca26c8 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java
@@ -20,8 +20,10 @@ package org.apache.samza.zk;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import org.apache.commons.lang3.StringUtils;
import org.apache.samza.config.ApplicationConfig;
import org.apache.samza.config.Config;
@@ -162,8 +164,14 @@ public class ZkJobCoordinator implements JobCoordinator, ZkControllerListener {
void doOnProcessorChange(List<String> processors) {
// if list of processors is empty - it means we are called from 'onBecomeLeader'
- // TODO: Handle empty currentProcessorIds or duplicate processorIds in the list
+ // TODO: Handle empty currentProcessorIds.
List<String> currentProcessorIds = getActualProcessorIds(processors);
+ Set<String> uniqueProcessorIds = new HashSet<String>(currentProcessorIds);
+
+ if (currentProcessorIds.size() != uniqueProcessorIds.size()) {
+ LOG.info("Processors: {} has duplicates. Not generating job model.", currentProcessorIds);
+ return;
+ }
// Generate the JobModel
JobModel jobModel = generateNewJobModel(currentProcessorIds);
http://git-wip-us.apache.org/repos/asf/samza/blob/45931fd5/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java b/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
index ecf118b..7406cf5 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
@@ -23,11 +23,13 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Objects;
+import java.util.TreeSet;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
-import org.I0Itec.zkclient.ZkConnection;
import org.I0Itec.zkclient.exception.ZkInterruptedException;
import org.apache.samza.SamzaException;
import org.apache.samza.job.model.JobModel;
@@ -83,10 +85,6 @@ public class ZkUtils {
}
}
- public static ZkConnection createZkConnection(String zkConnectString, int sessionTimeoutMs) {
- return new ZkConnection(zkConnectString, sessionTimeoutMs);
- }
-
ZkClient getZkClient() {
return zkClient;
}
@@ -105,17 +103,69 @@ public class ZkUtils {
* @return String representing the absolute ephemeralPath of this client in the current session
*/
public synchronized String registerProcessorAndGetId(final ProcessorData data) {
+ String processorId = data.getProcessorId();
if (ephemeralPath == null) {
- ephemeralPath =
- zkClient.createEphemeralSequential(
- keyBuilder.getProcessorsPath() + "/", data.toString());
-
- LOG.info("newly generated path for " + data + " is " + ephemeralPath);
- return ephemeralPath;
+ ephemeralPath = zkClient.createEphemeralSequential(keyBuilder.getProcessorsPath() + "/", data.toString());
+ LOG.info("Created ephemeral path: {} for processor: {} in zookeeper.", ephemeralPath, data);
+ ProcessorNode processorNode = new ProcessorNode(data, ephemeralPath);
+ // Determine if there are duplicate processors with this.processorId after registration.
+ if (!isValidRegisteredProcessor(processorNode)) {
+ LOG.info("Processor: {} is duplicate. Deleting zookeeper node at path: {}.", processorId, ephemeralPath);
+ zkClient.delete(ephemeralPath);
+ throw new SamzaException(String.format("Processor: %s is duplicate in the group. Registration failed.", processorId));
+ }
} else {
- LOG.info("existing path for " + data + " is " + ephemeralPath);
- return ephemeralPath;
+ LOG.info("Ephemeral path: {} exists for processor: {} in zookeeper.", ephemeralPath, data);
}
+ return ephemeralPath;
+ }
+
+ /**
+ * Determines the validity of processor registered with zookeeper.
+ *
+ * If there are multiple processors registered with same processorId,
+ * the processor with lexicographically smallest zookeeperPath is considered valid
+ * and all the remaining processors are invalid.
+ *
+ * Two processors will not have smallest zookeeperPath because of sequentialId guarantees
+ * of zookeeper for ephemeral nodes.
+ *
+ * @param processor to check for validity condition in processors group.
+ * @return true if the processor is valid. false otherwise.
+ */
+ private boolean isValidRegisteredProcessor(final ProcessorNode processor) {
+ String processorId = processor.getProcessorData().getProcessorId();
+ List<ProcessorNode> processorNodes = getAllProcessorNodes().stream()
+ .filter(processorNode -> processorNode.processorData.getProcessorId().equals(processorId))
+ .collect(Collectors.toList());
+ // Check for duplicate processor condition(if more than one processor exist for this processorId).
+ if (processorNodes.size() > 1) {
+ // There exists more than processor for provided `processorId`.
+ LOG.debug("Processor nodes in zookeeper: {} for processorId: {}.", processorNodes, processorId);
+ // Get all ephemeral processor paths
+ TreeSet<String> sortedProcessorPaths = processorNodes.stream()
+ .map(ProcessorNode::getEphemeralPath)
+ .collect(Collectors.toCollection(TreeSet::new));
+ // Check if smallest path is equal to this processor's ephemeralPath.
+ return sortedProcessorPaths.first().equals(processor.getEphemeralPath());
+ }
+ // There're no duplicate processors. This is a valid registered processor.
+ return true;
+ }
+
+ /**
+ * Fetches all the ephemeral processor nodes of a standalone job from zookeeper.
+ * @return a list of {@link ProcessorNode}, where each ProcessorNode represents a registered stream processor.
+ */
+ private List<ProcessorNode> getAllProcessorNodes() {
+ List<String> processorZNodes = getSortedActiveProcessorsZnodes();
+ LOG.debug("Active ProcessorZNodes in zookeeper: {}.", processorZNodes);
+ return processorZNodes.stream()
+ .map(processorZNode -> {
+ String ephemeralProcessorPath = String.format("%s/%s", keyBuilder.getProcessorsPath(), processorZNode);
+ String data = readProcessorData(ephemeralProcessorPath);
+ return new ProcessorNode(new ProcessorData(data), ephemeralProcessorPath);
+ }).collect(Collectors.toList());
}
/**
@@ -321,4 +371,45 @@ public class ZkUtils {
zkClient.subscribeChildChanges(keyBuilder.getProcessorsPath(), listener);
metrics.subscriptions.inc();
}
+
+ /**
+ * Represents zookeeper processor node.
+ */
+ private static class ProcessorNode {
+ private final ProcessorData processorData;
+
+ // Ex: /test/processors/0000000000
+ private final String ephemeralProcessorPath;
+
+ ProcessorNode(ProcessorData processorData, String ephemeralProcessorPath) {
+ this.processorData = processorData;
+ this.ephemeralProcessorPath = ephemeralProcessorPath;
+ }
+
+ ProcessorData getProcessorData() {
+ return processorData;
+ }
+
+ String getEphemeralPath() {
+ return ephemeralProcessorPath;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("[ProcessorData: %s, ephemeralProcessorPath: %s]", processorData, ephemeralProcessorPath);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(processorData, ephemeralProcessorPath);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == null) return false;
+ if (getClass() != obj.getClass()) return false;
+ final ProcessorNode other = (ProcessorNode) obj;
+ return Objects.equals(processorData, other.processorData) && Objects.equals(ephemeralProcessorPath, other.ephemeralProcessorPath);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/45931fd5/samza-core/src/test/java/org/apache/samza/zk/TestZkUtils.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/zk/TestZkUtils.java b/samza-core/src/test/java/org/apache/samza/zk/TestZkUtils.java
index a33bf03..e7a9aa2 100644
--- a/samza-core/src/test/java/org/apache/samza/zk/TestZkUtils.java
+++ b/samza-core/src/test/java/org/apache/samza/zk/TestZkUtils.java
@@ -37,6 +37,8 @@ import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.rules.ExpectedException;
import org.junit.Test;
public class TestZkUtils {
@@ -47,6 +49,10 @@ public class TestZkUtils {
private static final int CONNECTION_TIMEOUT_MS = 10000;
private ZkUtils zkUtils;
+ @Rule
+ // Declared public to honor junit contract.
+ public final ExpectedException expectedException = ExpectedException.none();
+
@BeforeClass
public static void setup() throws InterruptedException {
zkServer = new EmbeddedZookeeper();
@@ -68,10 +74,7 @@ public class TestZkUtils {
// Do nothing
}
- zkUtils = new ZkUtils(
- KEY_BUILDER,
- zkClient,
- SESSION_TIMEOUT_MS, new NoOpMetricsRegistry());
+ zkUtils = getZkUtils();
zkUtils.connect();
}
@@ -82,6 +85,11 @@ public class TestZkUtils {
zkClient.close();
}
+ private ZkUtils getZkUtils() {
+ return new ZkUtils(KEY_BUILDER, zkClient,
+ SESSION_TIMEOUT_MS, new NoOpMetricsRegistry());
+ }
+
@AfterClass
public static void teardown() {
zkServer.teardown();
@@ -174,6 +182,26 @@ public class TestZkUtils {
Assert.assertTrue(testWithDelayBackOff(() -> "newProcessor".equals(res.getRes()), 2, 1000));
}
+ /**
+ * Create two duplicate processors with same processorId.
+ * Second creation should fail with exception.
+ */
+ @Test
+ public void testRegisterProcessorAndGetIdShouldFailForDuplicateProcessorRegistration() {
+ final String testHostName = "localhost";
+ final String testProcessId = "testProcessorId";
+ ProcessorData processorData1 = new ProcessorData(testHostName, testProcessId);
+ // Register processor 1 which is not duplicate, this registration should succeed.
+ zkUtils.registerProcessorAndGetId(processorData1);
+
+ ZkUtils zkUtils1 = getZkUtils();
+ zkUtils1.connect();
+ ProcessorData duplicateProcessorData = new ProcessorData(testHostName, testProcessId);
+ // Registration of the duplicate processor should fail.
+ expectedException.expect(SamzaException.class);
+ zkUtils1.registerProcessorAndGetId(duplicateProcessorData);
+ }
+
@Test
public void testPublishNewJobModel() {
ZkKeyBuilder keyBuilder = new ZkKeyBuilder("test");
http://git-wip-us.apache.org/repos/asf/samza/blob/45931fd5/samza-test/src/test/java/org/apache/samza/test/processor/TestZkLocalApplicationRunner.java
----------------------------------------------------------------------
diff --git a/samza-test/src/test/java/org/apache/samza/test/processor/TestZkLocalApplicationRunner.java b/samza-test/src/test/java/org/apache/samza/test/processor/TestZkLocalApplicationRunner.java
index 2d5da2b..77e2a49 100644
--- a/samza-test/src/test/java/org/apache/samza/test/processor/TestZkLocalApplicationRunner.java
+++ b/samza-test/src/test/java/org/apache/samza/test/processor/TestZkLocalApplicationRunner.java
@@ -58,6 +58,7 @@ import org.apache.samza.zk.ZkKeyBuilder;
import org.apache.samza.zk.ZkUtils;
import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.ExpectedException;
import org.junit.rules.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -98,6 +99,9 @@ public class TestZkLocalApplicationRunner extends StandaloneIntegrationTestHarne
@Rule
public Timeout testTimeOutInMillis = new Timeout(90000);
+ @Rule
+ public final ExpectedException expectedException = ExpectedException.none();
+
@Override
public void setUp() {
super.setUp();
@@ -224,9 +228,7 @@ public class TestZkLocalApplicationRunner extends StandaloneIntegrationTestHarne
assertEquals("2", currentJobModelVersion);
}
- // Checks enforcing property that all processors should have unique Id.
- // Depends upon SAMZA-1302
- // @Test(expected = Exception.class)
+ @Test
public void shouldFailWhenNewProcessorJoinsWithSameIdAsExistingProcessor() throws InterruptedException {
// Set up kafka topics.
publishKafkaEvents(inputKafkaTopic, NUM_KAFKA_EVENTS, PROCESSOR_IDS[0]);
@@ -253,10 +255,9 @@ public class TestZkLocalApplicationRunner extends StandaloneIntegrationTestHarne
publishKafkaEvents(inputKafkaTopic, NUM_KAFKA_EVENTS, PROCESSOR_IDS[2]);
kafkaEventsConsumedLatch = new CountDownLatch(NUM_KAFKA_EVENTS);
StreamApplication streamApp3 = new TestStreamApplication(inputKafkaTopic, outputKafkaTopic, null, null, kafkaEventsConsumedLatch);
+ // Fail when the duplicate processor joins.
+ expectedException.expect(SamzaException.class);
applicationRunner3.run(streamApp3);
-
- // The following line should throw up by handling duplicate processorId registration.
- kafkaEventsConsumedLatch.await();
}
// Depends upon SAMZA-1302