Skip to content

SAMZA-2339 : Update KafkaSystemAdmin to adhere to the checkpoint replication factor #1177

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rmatharu-zz merged 1 commit into from
Oct 8, 2019
Merged

SAMZA-2339 : Update KafkaSystemAdmin to adhere to the checkpoint replication factor #1177

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions samza-api/src/main/java/org/apache/samza/system/StreamSpec.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -226,6 +226,10 @@ public boolean isCoordinatorStream() {
return id.equals(COORDINATOR_STREAM_ID);
}

public boolean isCheckpointStream() {
return id.equals(CHECKPOINT_STREAM_ID);
}

private void validateLogicalIdentifier(String identifierName, String identifierValue) {
if (identifierValue == null || !identifierValue.matches("[A-Za-z0-9_-]+")) {
throw new IllegalArgumentException(String.format("Identifier '%s' is '%s'. It must match the expression [A-Za-z0-9_-]+", identifierName, identifierValue));
Expand Down
3 changes: 3 additions & 0 deletions samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -540,6 +540,9 @@ public KafkaStreamSpec toKafkaSpec(StreamSpec spec) {
kafkaSpec =
new KafkaStreamSpec(spec.getId(), spec.getPhysicalName(), systemName, 1, coordinatorStreamReplicationFactor,
coordinatorStreamProperties);
} else if (spec.isCheckpointStream()) {
kafkaSpec = KafkaStreamSpec.fromSpec(StreamSpec.createCheckpointStreamSpec(spec.getPhysicalName(), systemName))
.copyWithReplicationFactor(Integer.parseInt(new KafkaConfig(config).getCheckpointReplicationFactor().get()));
Comment on lines +543 to +545

@rmatharu-zz rmatharu-zz Oct 3, 2019 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This makes sense, however, KafkaCheckpointManagerFactory already adheres to these configs when creating the spec.
Please see KafkaCheckpointManagerFactory Line 52-55.

val checkpointSpec = KafkaStreamSpec.fromSpec(StreamSpec.createCheckpointStreamSpec(checkpointTopic, checkpointSystemName))
.copyWithReplicationFactor(kafkaConfig.getCheckpointReplicationFactor.get.toInt)
.copyWithProperties(kafkaConfig.getCheckpointTopicProperties)

@VartulAgrawal VartulAgrawal Oct 3, 2019 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The KafkaCheckpointManagerFactory is only created after the KafkaSystemAdmin is initialised so by the time KafkaCheckpointManagerFactory is called the replicationFactor is already reset to 2 (https://github.com/apache/samza/blob/master/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java#L552 which is default for any general topic) and hence it needs to be passed to the spec here. Tested this in our current setup too :)

@rmatharu-zz rmatharu-zz Oct 8, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks. I'll merge your patch for now.
The problem really is that the KafkaSystemAdmin's toStreamSpec does not respect the incoming repl-factor in case the stream-spec is already a KafkaStreamSpec and tries to differentiate based on the stream-ID.

I've summarized the problem here https://issues.apache.org/jira/browse/SAMZA-2342
for a better, clean, long-term fix.

} else if (intermediateStreamProperties.containsKey(spec.getId())) {
kafkaSpec = KafkaStreamSpec.fromSpec(spec);
Properties properties = kafkaSpec.getProperties();
Expand Down
17 changes: 17 additions & 0 deletions samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemAdmin.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -271,6 +271,23 @@ class TestKafkaSystemAdmin {
validateTopic(topic, 1)
}

@Test
def testShouldCreateCheckpointStream {
val topic = "test-checkpoint-stream"
val expectedReplicationFactor = 1
val map = new java.util.HashMap[String, String]()
map.put(org.apache.samza.config.KafkaConfig.CHECKPOINT_REPLICATION_FACTOR, expectedReplicationFactor.toString)
val systemAdmin = createSystemAdmin(SYSTEM, map)

val spec = StreamSpec.createCheckpointStreamSpec(topic, "kafka")
val actualReplicationFactor = systemAdmin.toKafkaSpec(spec).getReplicationFactor
// Ensure we respect the replication factor passed to system admin
assertEquals(expectedReplicationFactor, actualReplicationFactor)

systemAdmin.createStream(spec)
validateTopic(topic, 1)
}

@Test
def testGetNewestOffset {
createTopic(TOPIC2, 16)
Expand Down