This is an automated email from the ASF dual-hosted git repository.
bharathkk pushed a commit to branch state-backend-async-commit
in repository https://gitbox.apache.org/repos/asf/samza.git
The following commit(s) were added to refs/heads/state-backend-async-commit by
this push:
new aa58950 SAMZA-2591: Async Commit [1/3]: Checkpoint v2 migration
(#1489)
aa58950 is described below
commit aa58950ee3b8487e9892d9638e893764cf3cc345
Author: Daniel Chen <[email protected]>
AuthorDate: Thu May 6 14:44:46 2021 -0700
SAMZA-2591: Async Commit [1/3]: Checkpoint v2 migration (#1489)
Introduce a new checkpoint message format (CheckpointV2) that includes
state markers (kafka changelog offset / ambry blob id) in them.
Renamed previous checkpoint messages as CheckpointV1
---
.../org/apache/samza/checkpoint/Checkpoint.java | 57 +++-----
.../org/apache/samza/checkpoint/CheckpointId.java | 51 ++++---
.../{Checkpoint.java => CheckpointV1.java} | 37 +++--
.../org/apache/samza/checkpoint/CheckpointV2.java | 123 +++++++++++++++++
.../samza/serializers/JsonCheckpointIdMixin.java | 42 ++++++
.../apache/samza/checkpoint/TestCheckpointId.java | 53 ++++++++
.../checkpoint/azure/AzureCheckpointManager.java | 10 +-
.../azure/ITestAzureCheckpointManager.java | 15 ++-
.../checkpoint/kafka/KafkaChangelogSSPOffset.java | 40 +++---
.../kafka/KafkaStateCheckpointMarker.java | 145 ++++++++++++++++++++
.../java/org/apache/samza/config/JobConfig.java | 28 ++++
.../org/apache/samza/config/StorageConfig.java | 44 ++++++
.../java/org/apache/samza/config/TaskConfig.java | 37 +++++
.../samza/serializers/CheckpointV2Serde.java | 59 ++++++++
.../apache/samza/serializers/JsonCheckpoint.java | 57 ++++++++
.../serializers/model/JsonCheckpointV2Mixin.java | 49 +++++++
.../model/KafkaStateCheckpointMarkerMixin.java | 48 +++++++
.../samza/serializers/model/SamzaObjectMapper.java | 23 ++++
.../apache/samza/checkpoint/CheckpointTool.scala | 5 +-
.../apache/samza/checkpoint/OffsetManager.scala | 30 +++--
.../file/FileSystemCheckpointManager.scala | 10 +-
...eckpointSerde.scala => CheckpointV1Serde.scala} | 69 +++-------
.../scala/org/apache/samza/util/FileUtil.scala | 9 +-
.../kafka/TestKafkaChangelogSSPOffset.java | 73 ++++++++++
.../kafka/TestKafkaStateCheckpointMarker.java | 114 ++++++++++++++++
.../samza/serializers/TestCheckpointV2Serde.java | 96 +++++++++++++
.../samza/checkpoint/TestCheckpointTool.scala | 12 +-
.../samza/checkpoint/TestOffsetManager.scala | 30 ++---
.../file/TestFileSystemCheckpointManager.scala | 10 +-
...ointSerde.scala => TestCheckpointV1Serde.scala} | 26 +---
.../scala/org/apache/samza/util/TestFileUtil.scala | 49 ++++++-
.../checkpoint/kafka/KafkaCheckpointLogKey.java | 8 +-
.../kafka/KafkaCheckpointLogKeySerde.java | 12 +-
.../checkpoint/kafka/KafkaCheckpointManager.scala | 105 ++++++++++-----
.../kafka/TestKafkaCheckpointLogKeySerde.java | 14 +-
.../kafka/TestKafkaCheckpointManagerJava.java | 29 ++--
.../kafka/TestKafkaCheckpointManager.scala | 129 ++++++++++++++++--
.../CheckpointVersionIntegrationTest.java | 149 +++++++++++++++++++++
.../kv/TransactionalStateIntegrationTest.java | 1 +
...ransactionalStateMultiStoreIntegrationTest.java | 1 +
40 files changed, 1616 insertions(+), 283 deletions(-)
diff --git
a/samza-api/src/main/java/org/apache/samza/checkpoint/Checkpoint.java
b/samza-api/src/main/java/org/apache/samza/checkpoint/Checkpoint.java
index 593d118..5f9276f 100644
--- a/samza-api/src/main/java/org/apache/samza/checkpoint/Checkpoint.java
+++ b/samza-api/src/main/java/org/apache/samza/checkpoint/Checkpoint.java
@@ -19,54 +19,27 @@
package org.apache.samza.checkpoint;
+import java.util.Map;
import org.apache.samza.system.SystemStreamPartition;
-import java.util.Collections;
-import java.util.Map;
-
-/**
- * A checkpoint is a mapping of all the streams a job is consuming and the
most recent current offset for each.
- * It is used to restore a {@link org.apache.samza.task.StreamTask}, either as
part of a job restart or as part
- * of restarting a failed container within a running job.
- */
-public class Checkpoint {
- private final Map<SystemStreamPartition, String> offsets;
+public interface Checkpoint {
/**
- * Constructs a new checkpoint based off a map of Samza stream offsets.
- * @param offsets Map of Samza streams to their current offset.
+ * Gets the version number of the Checkpoint
+ * @return Short indicating the version number
*/
- public Checkpoint(Map<SystemStreamPartition, String> offsets) {
- this.offsets = offsets;
- }
+ short getVersion();
/**
- * Gets a unmodifiable view of the current Samza stream offsets.
- * @return A unmodifiable view of a Map of Samza streams to their recorded
offsets.
+ * Gets a unmodifiable view of the last processed offsets for {@link
SystemStreamPartition}s.
+ * The returned value differs based on the Checkpoint version:
+ * <ol>
+ * <li>For {@link CheckpointV1}, returns the input {@link
SystemStreamPartition} offsets, as well
+ * as the latest KafkaStateChangelogOffset for any store changelog
{@link SystemStreamPartition} </li>
+ * <li>For {@link CheckpointV2} returns the input offsets only.</li>
+ * </ol>
+ *
+ * @return a unmodifiable view of last processed offsets for {@link
SystemStreamPartition}s.
*/
- public Map<SystemStreamPartition, String> getOffsets() {
- return Collections.unmodifiableMap(offsets);
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (!(o instanceof Checkpoint)) return false;
-
- Checkpoint that = (Checkpoint) o;
-
- if (offsets != null ? !offsets.equals(that.offsets) : that.offsets !=
null) return false;
-
- return true;
- }
-
- @Override
- public int hashCode() {
- return offsets != null ? offsets.hashCode() : 0;
- }
-
- @Override
- public String toString() {
- return "Checkpoint [offsets=" + offsets + "]";
- }
+ Map<SystemStreamPartition, String> getOffsets();
}
diff --git
a/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointId.java
b/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointId.java
index 95dfd24..f2aa8e1 100644
--- a/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointId.java
+++ b/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointId.java
@@ -16,54 +16,64 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.samza.checkpoint;
import java.util.Objects;
import org.apache.commons.lang3.StringUtils;
import org.apache.samza.annotation.InterfaceStability;
+
/**
* Checkpoint ID has the format: [currentTimeMillis, last 6 digits of
nanotime], separated by a dash.
* This is to avoid conflicts, e.g when requesting frequent manual commits.
*
- * It is expected that persistent stores use the {@link #toString()}
representation of the checkpoint id
+ * It is expected that persistent stores use the {@link #serialize()}
representation of the checkpoint id
* as the store checkpoint directory name.
*/
@InterfaceStability.Unstable
-public class CheckpointId {
+public class CheckpointId implements Comparable<CheckpointId> {
public static final String SEPARATOR = "-";
private final long millis;
- private final long nanos;
+ private final long nanoId;
- public CheckpointId(long millis, long nanos) {
+ private CheckpointId(long millis, long nanoId) {
this.millis = millis;
- this.nanos = nanos;
+ this.nanoId = nanoId;
}
public static CheckpointId create() {
return new CheckpointId(System.currentTimeMillis(), System.nanoTime() %
1000000);
}
- public static CheckpointId fromString(String checkpointId) {
+ public static CheckpointId deserialize(String checkpointId) {
if (StringUtils.isBlank(checkpointId)) {
throw new IllegalArgumentException("Invalid checkpoint id: " +
checkpointId);
}
- String[] parts = checkpointId.split(SEPARATOR);
- return new CheckpointId(Long.parseLong(parts[0]),
Long.parseLong(parts[1]));
+ try {
+ String[] parts = checkpointId.split(SEPARATOR);
+ return new CheckpointId(Long.parseLong(parts[0]),
Long.parseLong(parts[1]));
+ } catch (NumberFormatException ex) {
+ throw new IllegalArgumentException(String.format(
+ "Could not deserialize CheckpointId: %s", checkpointId), ex);
+ }
}
public long getMillis() {
return millis;
}
- public long getNanos() {
- return nanos;
+ public long getNanoId() {
+ return nanoId;
}
- @Override
- public String toString() {
- return String.format("%s%s%s", millis, SEPARATOR, nanos);
+ /**
+ * Serialization of {@link CheckpointId} as part of task checkpoints, in
conjunction with {@link #deserialize(String)}.
+ * @return the String representation of this {@link CheckpointId}.
+ */
+ public String serialize() {
+ return String.format("%s%s%s", getMillis(), SEPARATOR, getNanoId());
}
@Override
@@ -72,11 +82,22 @@ public class CheckpointId {
if (o == null || getClass() != o.getClass()) return false;
CheckpointId that = (CheckpointId) o;
return millis == that.millis &&
- nanos == that.nanos;
+ nanoId == that.nanoId;
}
@Override
public int hashCode() {
- return Objects.hash(millis, nanos);
+ return Objects.hash(millis, nanoId);
+ }
+
+ @Override
+ public int compareTo(CheckpointId that) {
+ if(this.millis != that.millis) return Long.compare(this.millis,
that.millis);
+ else return Long.compare(this.nanoId, that.nanoId);
+ }
+
+ @Override
+ public String toString() {
+ return String.format("%s%s%s", millis, SEPARATOR, nanoId);
}
}
\ No newline at end of file
diff --git
a/samza-api/src/main/java/org/apache/samza/checkpoint/Checkpoint.java
b/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointV1.java
similarity index 69%
copy from samza-api/src/main/java/org/apache/samza/checkpoint/Checkpoint.java
copy to samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointV1.java
index 593d118..3e2cce3 100644
--- a/samza-api/src/main/java/org/apache/samza/checkpoint/Checkpoint.java
+++ b/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointV1.java
@@ -23,50 +23,61 @@ import org.apache.samza.system.SystemStreamPartition;
import java.util.Collections;
import java.util.Map;
+import java.util.Objects;
/**
* A checkpoint is a mapping of all the streams a job is consuming and the
most recent current offset for each.
* It is used to restore a {@link org.apache.samza.task.StreamTask}, either as
part of a job restart or as part
* of restarting a failed container within a running job.
*/
-public class Checkpoint {
+public class CheckpointV1 implements Checkpoint {
+ public static final short CHECKPOINT_VERSION = 1;
+
private final Map<SystemStreamPartition, String> offsets;
/**
* Constructs a new checkpoint based off a map of Samza stream offsets.
+ *
* @param offsets Map of Samza streams to their current offset.
*/
- public Checkpoint(Map<SystemStreamPartition, String> offsets) {
+ public CheckpointV1(Map<SystemStreamPartition, String> offsets) {
this.offsets = offsets;
}
+ public short getVersion() {
+ return CHECKPOINT_VERSION;
+ }
+
/**
- * Gets a unmodifiable view of the current Samza stream offsets.
- * @return A unmodifiable view of a Map of Samza streams to their recorded
offsets.
+ * Gets a unmodifiable view of the current Samza input stream offsets.
+ *
+ * @return a unmodifiable view of a Map of Samza streams to their recorded
offsets.
*/
+ @Override
public Map<SystemStreamPartition, String> getOffsets() {
return Collections.unmodifiableMap(offsets);
}
@Override
public boolean equals(Object o) {
- if (this == o) return true;
- if (!(o instanceof Checkpoint)) return false;
-
- Checkpoint that = (Checkpoint) o;
-
- if (offsets != null ? !offsets.equals(that.offsets) : that.offsets !=
null) return false;
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ CheckpointV1 that = (CheckpointV1) o;
- return true;
+ return Objects.equals(offsets, that.offsets);
}
@Override
public int hashCode() {
- return offsets != null ? offsets.hashCode() : 0;
+ return Objects.hash(offsets);
}
@Override
public String toString() {
- return "Checkpoint [offsets=" + offsets + "]";
+ return "CheckpointV1 [CHECKPOINT_VERSION=" + CHECKPOINT_VERSION + ",
offsets=" + offsets + "]";
}
}
diff --git
a/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointV2.java
b/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointV2.java
new file mode 100644
index 0000000..ac553f4
--- /dev/null
+++ b/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointV2.java
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.checkpoint;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import java.util.Objects;
+import org.apache.samza.system.SystemStreamPartition;
+
+import java.util.Map;
+
+/**
+ * A checkpoint is a mapping of all the streams a job is consuming and the
most recent current offset for each.
+ * It is used to restore a {@link org.apache.samza.task.StreamTask}, either as
part of a job restart or as part
+ * of restarting a failed container within a running job.
+ */
+
+public class CheckpointV2 implements Checkpoint {
+ public static final short CHECKPOINT_VERSION = 2;
+
+ private final CheckpointId checkpointId;
+ private final Map<SystemStreamPartition, String> inputOffsets;
+ private final Map<String, Map<String, String>> stateCheckpointMarkers;
+
+ /**
+ * Constructs the checkpoint with separated input and state offsets
+ *
+ * @param checkpointId {@link CheckpointId} associated with this checkpoint
+ * @param inputOffsets Map of Samza system stream partition to offset of the
checkpoint
+ * @param stateCheckpoints Map of state backend factory name to map of local
state store names
+ * to state checkpoints
+ */
+ public CheckpointV2(CheckpointId checkpointId,
+ Map<SystemStreamPartition, String> inputOffsets,
+ Map<String, Map<String, String>> stateCheckpoints) {
+ Preconditions.checkArgument(inputOffsets != null,
+ "inputOffsets for CheckpointV2 must not be null");
+ Preconditions.checkArgument(stateCheckpoints != null,
+ "stateCheckpoints for CheckpointV2 must not be null");
+ this.checkpointId = checkpointId;
+ this.inputOffsets = ImmutableMap.copyOf(inputOffsets);
+ this.stateCheckpointMarkers = ImmutableMap.copyOf(stateCheckpoints);
+ }
+
+ public short getVersion() {
+ return CHECKPOINT_VERSION;
+ }
+
+ /**
+ * Gets the checkpoint id for the checkpoint
+ * @return The timestamp based checkpoint identifier associated with the
checkpoint
+ */
+ public CheckpointId getCheckpointId() {
+ return checkpointId;
+ }
+
+ /**
+ * Gets a unmodifiable view of the current input {@link
SystemStreamPartition} offsets.
+ * @return An unmodifiable map of input {@link SystemStreamPartition}s to
their recorded offsets.
+ */
+ @Override
+ public Map<SystemStreamPartition, String> getOffsets() {
+ return inputOffsets;
+ }
+
+ /**
+ * Gets the state checkpoint markers for all stores for each configured
state backend.
+ *
+ * Note: We don't add this method to the {@link Checkpoint} interface since
it is difficult
+ * to implement it for {@link CheckpointV1} without changing the underlying
serialization format -
+ * the changelog SSP offsets are serialized in the same way as input
offsets, and at
+ * deserialization time we don't have enough information (e.g. configs) to
decide whether a
+ * particular entry is for an input SSP or a changelog SSP.
+ *
+ * @return Map of state backend factory name to map of local state store
names to state checkpoint markers
+ */
+ public Map<String, Map<String, String>> getStateCheckpointMarkers() {
+ return stateCheckpointMarkers;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ CheckpointV2 that = (CheckpointV2) o;
+
+ return checkpointId.equals(that.checkpointId) &&
+ Objects.equals(inputOffsets, that.inputOffsets) &&
+ Objects.equals(stateCheckpointMarkers, that.stateCheckpointMarkers);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(checkpointId, inputOffsets, stateCheckpointMarkers);
+ }
+
+ @Override
+ public String toString() {
+ return "CheckpointV2 [CHECKPOINT_VERSION=" + CHECKPOINT_VERSION + ",
checkpointId=" + checkpointId +
+ ", inputOffsets=" + inputOffsets + ", stateCheckpointMarkers=" +
stateCheckpointMarkers + "]";
+ }
+}
\ No newline at end of file
diff --git
a/samza-api/src/main/java/org/apache/samza/serializers/JsonCheckpointIdMixin.java
b/samza-api/src/main/java/org/apache/samza/serializers/JsonCheckpointIdMixin.java
new file mode 100644
index 0000000..752e329
--- /dev/null
+++
b/samza-api/src/main/java/org/apache/samza/serializers/JsonCheckpointIdMixin.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.serializers;
+
+import org.apache.samza.checkpoint.CheckpointId;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+
+/**
+ * A mix-in Jackson class to convert {@link CheckpointId} to/from JSON.
+ */
+public abstract class JsonCheckpointIdMixin {
+ @JsonCreator
+ private JsonCheckpointIdMixin(
+ @JsonProperty("millis") long millis,
+ @JsonProperty("nanos") long nanos) {
+ }
+
+ @JsonProperty("millis")
+ abstract long getMillis();
+
+ @JsonProperty("nanos")
+ abstract long getNanos();
+}
diff --git
a/samza-api/src/test/java/org/apache/samza/checkpoint/TestCheckpointId.java
b/samza-api/src/test/java/org/apache/samza/checkpoint/TestCheckpointId.java
new file mode 100644
index 0000000..2ac1b88
--- /dev/null
+++ b/samza-api/src/test/java/org/apache/samza/checkpoint/TestCheckpointId.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.checkpoint;
+
+import org.junit.Test;
+
+import static org.apache.samza.checkpoint.CheckpointId.*;
+import static org.junit.Assert.assertEquals;
+
+
+public class TestCheckpointId {
+ @Test
+ public void testSerializationDeserialization() {
+ CheckpointId checkpointId = CheckpointId.create();
+ CheckpointId deserializedCheckpointId =
CheckpointId.deserialize(checkpointId.serialize());
+
+ assertEquals(checkpointId.getMillis(),
deserializedCheckpointId.getMillis());
+ assertEquals(checkpointId.getNanoId(),
deserializedCheckpointId.getNanoId());
+ assertEquals(checkpointId, deserializedCheckpointId);
+ }
+
+ @Test
+ public void testSerializationFormatForBackwardsCompatibility() {
+ CheckpointId checkpointId = CheckpointId.create();
+ String serializedCheckpointId = checkpointId.serialize();
+
+ // WARNING: This format is written to persisted remotes stores and local
files, making a change in the format
+ // would be backwards incompatible
+ String legacySerializedFormat = serializeLegacy(checkpointId);
+ assertEquals(checkpointId,
CheckpointId.deserialize(legacySerializedFormat));
+ }
+
+ public String serializeLegacy(CheckpointId id) {
+ return String.format("%s%s%s", id.getMillis(), SEPARATOR, id.getNanoId());
+ }
+}
diff --git
a/samza-azure/src/main/java/org/apache/samza/checkpoint/azure/AzureCheckpointManager.java
b/samza-azure/src/main/java/org/apache/samza/checkpoint/azure/AzureCheckpointManager.java
index 9013196..1d49c2c 100644
---
a/samza-azure/src/main/java/org/apache/samza/checkpoint/azure/AzureCheckpointManager.java
+++
b/samza-azure/src/main/java/org/apache/samza/checkpoint/azure/AzureCheckpointManager.java
@@ -19,6 +19,7 @@
package org.apache.samza.checkpoint.azure;
+import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.microsoft.azure.storage.StorageException;
import com.microsoft.azure.storage.table.*;
@@ -28,6 +29,7 @@ import org.apache.samza.Partition;
import org.apache.samza.SamzaException;
import org.apache.samza.checkpoint.Checkpoint;
import org.apache.samza.checkpoint.CheckpointManager;
+import org.apache.samza.checkpoint.CheckpointV1;
import org.apache.samza.config.AzureConfig;
import org.apache.samza.container.TaskName;
import org.apache.samza.serializers.JsonSerdeV2;
@@ -113,13 +115,17 @@ public class AzureCheckpointManager implements
CheckpointManager {
@Override
public void writeCheckpoint(TaskName taskName, Checkpoint checkpoint) {
+ Preconditions.checkArgument(checkpoint instanceof CheckpointV1,
+ "Only CheckpointV1 could be written to Azure");
+
if (!taskNames.contains(taskName)) {
throw new SamzaException("writing checkpoint of unregistered task");
}
TableBatchOperation batchOperation = new TableBatchOperation();
- Iterator<Map.Entry<SystemStreamPartition, String>> iterator =
checkpoint.getOffsets().entrySet().iterator();
+ Iterator<Map.Entry<SystemStreamPartition, String>> iterator =
+ checkpoint.getOffsets().entrySet().iterator();
while (iterator.hasNext()) {
Map.Entry<SystemStreamPartition, String> entry = iterator.next();
SystemStreamPartition ssp = entry.getKey();
@@ -205,7 +211,7 @@ public class AzureCheckpointManager implements
CheckpointManager {
return null;
}
LOG.debug("Received checkpoint state for taskName=%s", taskName);
- return new Checkpoint(builder.build());
+ return new CheckpointV1(builder.build());
}
@Override
diff --git
a/samza-azure/src/test/java/org/apache/samza/checkpoint/azure/ITestAzureCheckpointManager.java
b/samza-azure/src/test/java/org/apache/samza/checkpoint/azure/ITestAzureCheckpointManager.java
index 7cf01c0..2240e65 100644
---
a/samza-azure/src/test/java/org/apache/samza/checkpoint/azure/ITestAzureCheckpointManager.java
+++
b/samza-azure/src/test/java/org/apache/samza/checkpoint/azure/ITestAzureCheckpointManager.java
@@ -22,6 +22,7 @@ package org.apache.samza.checkpoint.azure;
import org.apache.samza.Partition;
import org.apache.samza.checkpoint.Checkpoint;
import org.apache.samza.checkpoint.CheckpointManager;
+import org.apache.samza.checkpoint.CheckpointV1;
import org.apache.samza.config.AzureConfig;
import org.apache.samza.config.Config;
import org.apache.samza.config.MapConfig;
@@ -69,10 +70,10 @@ public class ITestAzureCheckpointManager {
Map<SystemStreamPartition, String> sspMap = new HashMap<>();
sspMap.put(ssp, "12345");
- Checkpoint cp0 = new Checkpoint(sspMap);
+ Checkpoint cp0 = new CheckpointV1(sspMap);
sspMap.put(ssp, "54321");
- Checkpoint cp1 = new Checkpoint(sspMap);
+ Checkpoint cp1 = new CheckpointV1(sspMap);
checkpointManager.register(taskName);
@@ -96,12 +97,12 @@ public class ITestAzureCheckpointManager {
Map<SystemStreamPartition, String> sspMap = new HashMap<>();
sspMap.put(ssp, "12345");
sspMap.put(ssp1, "54321");
- Checkpoint cp1 = new Checkpoint(sspMap);
+ Checkpoint cp1 = new CheckpointV1(sspMap);
Map<SystemStreamPartition, String> sspMap2 = new HashMap<>();
sspMap2.put(ssp, "12347");
sspMap2.put(ssp1, "54323");
- Checkpoint cp2 = new Checkpoint(sspMap2);
+ Checkpoint cp2 = new CheckpointV1(sspMap2);
checkpointManager.register(taskName);
@@ -126,12 +127,12 @@ public class ITestAzureCheckpointManager {
Map<SystemStreamPartition, String> sspMap = new HashMap<>();
sspMap.put(ssp, "12345");
sspMap.put(ssp1, "54321");
- Checkpoint cp1 = new Checkpoint(sspMap);
+ Checkpoint cp1 = new CheckpointV1(sspMap);
Map<SystemStreamPartition, String> sspMap2 = new HashMap<>();
sspMap2.put(ssp, "12347");
sspMap2.put(ssp1, "54323");
- Checkpoint cp2 = new Checkpoint(sspMap2);
+ Checkpoint cp2 = new CheckpointV1(sspMap2);
checkpointManager.register(taskName);
checkpointManager.register(taskName1);
@@ -171,7 +172,7 @@ public class ITestAzureCheckpointManager {
sspMap.put(ssp, String.valueOf(i));
}
- Checkpoint cp0 = new Checkpoint(sspMap);
+ Checkpoint cp0 = new CheckpointV1(sspMap);
checkpointManager.register(taskName);
checkpointManager.writeCheckpoint(taskName, cp0);
Checkpoint readCp = checkpointManager.readLastCheckpoint(taskName);
diff --git
a/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointedChangelogOffset.java
b/samza-core/src/main/java/org/apache/samza/checkpoint/kafka/KafkaChangelogSSPOffset.java
similarity index 57%
rename from
samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointedChangelogOffset.java
rename to
samza-core/src/main/java/org/apache/samza/checkpoint/kafka/KafkaChangelogSSPOffset.java
index 407ce7a..09ea497 100644
---
a/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointedChangelogOffset.java
+++
b/samza-core/src/main/java/org/apache/samza/checkpoint/kafka/KafkaChangelogSSPOffset.java
@@ -16,67 +16,75 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.samza.checkpoint;
+package org.apache.samza.checkpoint.kafka;
import java.util.Objects;
import org.apache.commons.lang3.StringUtils;
import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.checkpoint.CheckpointId;
/**
- * Checkpointed changelog offset has the format: [checkpointId, offset],
separated by a colon.
+ * Used in {@link org.apache.samza.checkpoint.CheckpointV1} for tracking the
latest offset for store changelogs at
+ * the time of commit. Checkpointed changelog offset has the format:
[checkpointId, offset], separated by a colon.
*/
@InterfaceStability.Unstable
-public class CheckpointedChangelogOffset {
+public class KafkaChangelogSSPOffset {
public static final String SEPARATOR = ":";
private final CheckpointId checkpointId;
- private final String offset;
+ private final String changelogOffset;
- public CheckpointedChangelogOffset(CheckpointId checkpointId, String offset)
{
+ public KafkaChangelogSSPOffset(CheckpointId checkpointId, String
changelogOffset) {
this.checkpointId = checkpointId;
- this.offset = offset;
+ this.changelogOffset = changelogOffset;
}
- public static CheckpointedChangelogOffset fromString(String message) {
+ public static KafkaChangelogSSPOffset fromString(String message) {
if (StringUtils.isBlank(message)) {
throw new IllegalArgumentException("Invalid checkpointed changelog
message: " + message);
}
- String[] checkpointIdAndOffset = message.split(":");
+ String[] checkpointIdAndOffset = message.split(SEPARATOR);
if (checkpointIdAndOffset.length != 2) {
throw new IllegalArgumentException("Invalid checkpointed changelog
offset: " + message);
}
- CheckpointId checkpointId =
CheckpointId.fromString(checkpointIdAndOffset[0]);
+ CheckpointId checkpointId =
CheckpointId.deserialize(checkpointIdAndOffset[0]);
String offset = null;
if (!"null".equals(checkpointIdAndOffset[1])) {
offset = checkpointIdAndOffset[1];
}
- return new CheckpointedChangelogOffset(checkpointId, offset);
+
+ return new KafkaChangelogSSPOffset(checkpointId, offset);
}
public CheckpointId getCheckpointId() {
return checkpointId;
}
- public String getOffset() {
- return offset;
+ public String getChangelogOffset() {
+ return changelogOffset;
}
+ /**
+ * WARNING: Do not change the toString() representation. It is used for
serde'ing the store changelog offsets
+ * as part of task checkpoints, in conjunction with {@link
#fromString(String)}.
+ * @return the String representation of this {@link KafkaChangelogSSPOffset}
+ */
@Override
public String toString() {
- return String.format("%s%s%s", checkpointId, SEPARATOR, offset);
+ return String.format("%s%s%s", checkpointId, SEPARATOR, changelogOffset);
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
- CheckpointedChangelogOffset that = (CheckpointedChangelogOffset) o;
+ KafkaChangelogSSPOffset that = (KafkaChangelogSSPOffset) o;
return Objects.equals(checkpointId, that.checkpointId) &&
- Objects.equals(offset, that.offset);
+ Objects.equals(changelogOffset, that.changelogOffset);
}
@Override
public int hashCode() {
- return Objects.hash(checkpointId, offset);
+ return Objects.hash(checkpointId, changelogOffset);
}
}
diff --git
a/samza-core/src/main/java/org/apache/samza/checkpoint/kafka/KafkaStateCheckpointMarker.java
b/samza-core/src/main/java/org/apache/samza/checkpoint/kafka/KafkaStateCheckpointMarker.java
new file mode 100644
index 0000000..ccd06d1
--- /dev/null
+++
b/samza-core/src/main/java/org/apache/samza/checkpoint/kafka/KafkaStateCheckpointMarker.java
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.checkpoint.kafka;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.samza.SamzaException;
+import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.serializers.model.SamzaObjectMapper;
+import org.apache.samza.storage.KafkaChangelogStateBackendFactory;
+import org.apache.samza.system.SystemStreamPartition;
+import scala.Option;
+
+
+/**
+ * Used as the serialization format for the state checkpoints of {@link
org.apache.samza.checkpoint.CheckpointV2}
+ * for a store using {@link
org.apache.samza.storage.KafkaTransactionalStateTaskBackupManager} or
+ * {@link
org.apache.samza.storage.KafkaNonTransactionalStateTaskBackupManager} for
tracking the latest committed
+ * store changelog offsets.
+ *
+ * Kafka state checkpoint marker has the format: [system, stream, partition,
offset], separated by a semi-colon.
+ */
[email protected]
+public class KafkaStateCheckpointMarker {
+ public static final String KAFKA_STATE_BACKEND_FACTORY_NAME =
KafkaChangelogStateBackendFactory.class.getName();
+ public static final short MARKER_VERSION = 1;
+ private static final ObjectMapper MAPPER =
SamzaObjectMapper.getObjectMapper();
+
+ // Required for Jackson Serde
+ private final short version;
+ private final SystemStreamPartition changelogSSP;
+ private final String changelogOffset;
+
+ public KafkaStateCheckpointMarker(SystemStreamPartition changelogSSP, String
changelogOffset) {
+ this(MARKER_VERSION, changelogSSP, changelogOffset);
+ }
+
+ public KafkaStateCheckpointMarker(short version, SystemStreamPartition
changelogSSP, String changelogOffset) {
+ this.version = version;
+ this.changelogSSP = changelogSSP;
+ this.changelogOffset = changelogOffset;
+ }
+
+ public static KafkaStateCheckpointMarker deserialize(String
stateCheckpointMarker) {
+ try {
+ return MAPPER.readValue(stateCheckpointMarker,
KafkaStateCheckpointMarker.class);
+ } catch (JsonProcessingException e) {
+ throw new IllegalArgumentException("Could not deserialize
KafkaStateCheckpointMarker: " + stateCheckpointMarker);
+ }
+ }
+
+ public short getVersion() {
+ return version;
+ }
+
+ public SystemStreamPartition getChangelogSSP() {
+ return changelogSSP;
+ }
+
+ public String getChangelogOffset() {
+ return changelogOffset;
+ }
+
+ /**
+ * Builds a map of store changelog SSPs to their offset for Kafka changelog
backed stores from the provided
+ * map of state backend factory name to map of store name to serialized
state checkpoint markers.
+ *
+ * @param stateBackendToStoreSCMs Map of state backend factory name to map
of store name to serialized
+ * state checkpoint markers
+ * @return Map of store changelog SSPss to their optional offset, or an
empty map if there is no mapping for
+ * {@link #KAFKA_STATE_BACKEND_FACTORY_NAME} in the input map. Optional
offset may be empty if the
+ * changelog SSP was empty.
+ */
+ public static Map<SystemStreamPartition, Option<String>> scmsToSSPOffsetMap(
+ Map<String, Map<String, String>> stateBackendToStoreSCMs) {
+ Map<SystemStreamPartition, Option<String>> sspToOffsetOptions = new
HashMap<>();
+ if (stateBackendToStoreSCMs.containsKey(KAFKA_STATE_BACKEND_FACTORY_NAME))
{
+ Map<String, String> storeToKafkaSCMs =
stateBackendToStoreSCMs.get(KAFKA_STATE_BACKEND_FACTORY_NAME);
+ storeToKafkaSCMs.forEach((key, value) -> {
+ KafkaStateCheckpointMarker stateMarker =
KafkaStateCheckpointMarker.deserialize(value);
+ Option<String> offsetOption =
Option.apply(stateMarker.getChangelogOffset());
+ sspToOffsetOptions.put(new
SystemStreamPartition(stateMarker.getChangelogSSP()), offsetOption);
+ });
+ }
+ return sspToOffsetOptions;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ KafkaStateCheckpointMarker that = (KafkaStateCheckpointMarker) o;
+ return Objects.equals(changelogSSP, that.changelogSSP) &&
+ Objects.equals(changelogOffset, that.changelogOffset);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(changelogSSP, changelogOffset);
+ }
+
+ /**
+ * It is used for serde'ing {@link KafkaStateCheckpointMarker}s, in
conjunction with {@link #deserialize(String)}.
+ * @return the String representation of this {@link
KafkaStateCheckpointMarker}
+ */
+ public static String serialize(KafkaStateCheckpointMarker marker) {
+ try {
+ return MAPPER.writeValueAsString(marker);
+ } catch (JsonProcessingException e) {
+ throw new SamzaException(String.format("Error serializing
KafkaCheckpointMarker %s", marker), e);
+ }
+ }
+
+ @Override
+ public String toString() {
+ String separator = ",";
+ return String.format("%s%s%s%s%s%s%s",
+ changelogSSP.getSystem(), separator, changelogSSP.getStream(),
separator,
+ changelogSSP.getPartition().getPartitionId(), separator,
changelogOffset);
+ }
+}
diff --git a/samza-core/src/main/java/org/apache/samza/config/JobConfig.java
b/samza-core/src/main/java/org/apache/samza/config/JobConfig.java
index 021c67d..12257e0 100644
--- a/samza-core/src/main/java/org/apache/samza/config/JobConfig.java
+++ b/samza-core/src/main/java/org/apache/samza/config/JobConfig.java
@@ -63,6 +63,18 @@ public class JobConfig extends MapConfig {
public static final String JOB_CONTAINER_COUNT = "job.container.count";
static final int DEFAULT_JOB_CONTAINER_COUNT = 1;
public static final String JOB_CONTAINER_THREAD_POOL_SIZE =
"job.container.thread.pool.size";
+ // num commit threads == min(max(2 * num tasks in container, thread pool
size), max thread pool size)
+ public static final String COMMIT_THREAD_POOL_SIZE =
"job.container.commit.thread.pool.size";
+ static final int DEFAULT_COMMIT_THREAD_POOL_SIZE = 2;
+ public static final String COMMIT_THREAD_POOL_MAX_SIZE =
"job.container.commit.thread.pool.max.size";
+ static final int DEFAULT_COMMIT_THREAD_POOL_MAX_SIZE = 64;
+
+ // num commit threads == min(max(2 * num tasks in container, thread pool
size), max thread pool size)
+ public static final String RESTORE_THREAD_POOL_SIZE =
"job.container.restore.thread.pool.size";
+ static final int DEFAULT_RESTORE_THREAD_POOL_SIZE = 2;
+ public static final String RESTORE_THREAD_POOL_MAX_SIZE =
"job.container.restore.thread.pool.max.size";
+ static final int DEFAULT_RESTORE_THREAD_POOL_MAX_SIZE = 64;
+
public static final String JOB_INTERMEDIATE_STREAM_PARTITIONS =
"job.intermediate.stream.partitions";
public static final String JOB_DEBOUNCE_TIME_MS = "job.debounce.time.ms";
@@ -336,6 +348,22 @@ public class JobConfig extends MapConfig {
}
}
+ public int getCommitThreadPoolSize() {
+ return getInt(COMMIT_THREAD_POOL_SIZE, DEFAULT_COMMIT_THREAD_POOL_SIZE);
+ }
+
+ public int getCommitThreadPoolMaxSize() {
+ return getInt(COMMIT_THREAD_POOL_MAX_SIZE,
DEFAULT_COMMIT_THREAD_POOL_MAX_SIZE);
+ }
+
+ public int getRestoreThreadPoolSize() {
+ return getInt(RESTORE_THREAD_POOL_SIZE, DEFAULT_RESTORE_THREAD_POOL_SIZE);
+ }
+
+ public int getRestoreThreadPoolMaxSize() {
+ return getInt(RESTORE_THREAD_POOL_MAX_SIZE,
DEFAULT_RESTORE_THREAD_POOL_MAX_SIZE);
+ }
+
public int getDebounceTimeMs() {
return getInt(JOB_DEBOUNCE_TIME_MS, DEFAULT_DEBOUNCE_TIME_MS);
}
diff --git
a/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java
b/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java
index a8b8702..f5e2055 100644
--- a/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java
+++ b/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java
@@ -19,16 +19,22 @@
package org.apache.samza.config;
+import com.google.common.collect.ImmutableList;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.commons.lang3.StringUtils;
import org.apache.samza.SamzaException;
import org.apache.samza.execution.StreamManager;
+import org.apache.samza.system.SystemStream;
import org.apache.samza.util.StreamUtil;
import static com.google.common.base.Preconditions.*;
@@ -61,6 +67,12 @@ public class StorageConfig extends MapConfig {
public static final String CHANGELOG_MIN_COMPACTION_LAG_MS = STORE_PREFIX +
"%s.changelog." + MIN_COMPACTION_LAG_MS;
public static final long DEFAULT_CHANGELOG_MIN_COMPACTION_LAG_MS =
TimeUnit.HOURS.toMillis(4);
+ public static final String DEFAULT_STATE_BACKEND_FACTORY =
"org.apache.samza.storage.KafkaChangelogStateBackendFactory";
+ public static final String STORE_BACKEND_BACKUP_FACTORIES = STORE_PREFIX +
"%s.state.backend.backup.factories";
+ public static final List<String> DEFAULT_STATE_BACKEND_BACKUP_FACTORIES =
ImmutableList.of(
+ DEFAULT_STATE_BACKEND_FACTORY);
+ public static final String STATE_BACKEND_RESTORE_FACTORY = STORE_PREFIX +
"state.restore.backend";
+
static final String CHANGELOG_SYSTEM = "job.changelog.system";
static final String CHANGELOG_DELETE_RETENTION_MS = STORE_PREFIX +
"%s.changelog.delete.retention.ms";
static final long DEFAULT_CHANGELOG_DELETE_RETENTION_MS =
TimeUnit.DAYS.toMillis(1);
@@ -94,6 +106,11 @@ public class StorageConfig extends MapConfig {
return storeNames;
}
+ public Map<String, SystemStream> getStoreChangelogs() {
+ return getStoreNames().stream().filter(store ->
getChangelogStream(store).isPresent())
+ .collect(Collectors.toMap(Function.identity(), n ->
StreamUtil.getSystemStreamFromNames(getChangelogStream(n).get())));
+ }
+
/**
* If the config specifies 'stores.<storename>.changelog' as
'<system>.<stream>' combination - it will take
* precedence.
@@ -122,6 +139,15 @@ public class StorageConfig extends MapConfig {
return Optional.ofNullable(systemStreamRes);
}
+ public List<String> getStoreBackupManagerClassName(String storeName) {
+ List<String> storeBackupManagers =
getList(String.format(STORE_BACKEND_BACKUP_FACTORIES, storeName), new
ArrayList<>());
+ // For backwards compatibility if the changelog is enabled, we use default
kafka backup factory
+ if (storeBackupManagers.isEmpty() &&
getChangelogStream(storeName).isPresent()) {
+ storeBackupManagers = DEFAULT_STATE_BACKEND_BACKUP_FACTORIES;
+ }
+ return storeBackupManagers;
+ }
+
public boolean getAccessLogEnabled(String storeName) {
return getBoolean(String.format(ACCESSLOG_ENABLED, storeName), false);
}
@@ -240,6 +266,24 @@ public class StorageConfig extends MapConfig {
return getLong(minCompactLagConfigName,
getDefaultChangelogMinCompactionLagMs());
}
+
+ public Set<String> getStateBackendBackupFactories() {
+ return getStoreNames().stream()
+ .flatMap((storeName) ->
getStoreBackupManagerClassName(storeName).stream())
+ .collect(Collectors.toSet());
+ }
+
+ public List<String> getBackupStoreNamesForStateBackupFactory(String
backendFactoryName) {
+ return getStoreNames().stream()
+ .filter((storeName) -> getStoreBackupManagerClassName(storeName)
+ .contains(backendFactoryName))
+ .collect(Collectors.toList());
+ }
+
+ public String getStateBackendRestoreFactory() {
+ return get(STATE_BACKEND_RESTORE_FACTORY, DEFAULT_STATE_BACKEND_FACTORY);
+ }
+
/**
* Helper method to check if a system has a changelog attached to it.
*/
diff --git a/samza-core/src/main/java/org/apache/samza/config/TaskConfig.java
b/samza-core/src/main/java/org/apache/samza/config/TaskConfig.java
index 461b647..8db1d2a 100644
--- a/samza-core/src/main/java/org/apache/samza/config/TaskConfig.java
+++ b/samza-core/src/main/java/org/apache/samza/config/TaskConfig.java
@@ -19,6 +19,9 @@
package org.apache.samza.config;
+import com.google.common.collect.ImmutableList;
+
+import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
@@ -53,6 +56,15 @@ public class TaskConfig extends MapConfig {
// commit period in milliseconds
public static final String COMMIT_MS = "task.commit.ms";
static final long DEFAULT_COMMIT_MS = 60000L;
+ // maximum amount of time a task may continue processing while a previous
commit is pending
+ public static final String COMMIT_MAX_DELAY_MS = "task.commit.max.delay.ms";
+ static final long DEFAULT_COMMIT_MAX_DELAY_MS =
Duration.ofMinutes(1).toMillis();
+ // maximum amount of time to block for a pending task commit to complete
*after*
+ // COMMIT_MAX_DELAY_MS have passed since the pending commit start. if the
pending commit
+ // does not complete within this timeout, the container will shut down.
+ public static final String COMMIT_TIMEOUT_MS = "task.commit.timeout.ms";
+ static final long DEFAULT_COMMIT_TIMEOUT_MS =
Duration.ofMinutes(1).toMillis();
+
// how long to wait for a clean shutdown
public static final String TASK_SHUTDOWN_MS = "task.shutdown.ms";
static final long DEFAULT_TASK_SHUTDOWN_MS = 30000L;
@@ -108,6 +120,14 @@ public class TaskConfig extends MapConfig {
// standby containers use this flag to indicate that checkpoints will be
polled continually, rather than only once at startup like in an active container
public static final String
INTERNAL_CHECKPOINT_MANAGER_CONSUMER_STOP_AFTER_FIRST_READ =
"samza.internal.task.checkpoint.consumer.stop.after.first.read";
+ // list of checkpoint versions to write during processing
+ public static final String CHECKPOINT_WRITE_VERSIONS =
"task.checkpoint.write.versions";
+ public static final List<String> DEFAULT_CHECKPOINT_WRITE_VERSIONS =
ImmutableList.of("1", "2");
+
+ // checkpoint version to read during container startup
+ public static final String CHECKPOINT_READ_VERSION =
"task.checkpoint.read.version";
+ public static final short DEFAULT_CHECKPOINT_READ_VERSION = 1;
+
public static final String TRANSACTIONAL_STATE_CHECKPOINT_ENABLED =
"task.transactional.state.checkpoint.enabled";
private static final boolean DEFAULT_TRANSACTIONAL_STATE_CHECKPOINT_ENABLED
= true;
public static final String TRANSACTIONAL_STATE_RESTORE_ENABLED =
"task.transactional.state.restore.enabled";
@@ -143,6 +163,14 @@ public class TaskConfig extends MapConfig {
return getLong(COMMIT_MS, DEFAULT_COMMIT_MS);
}
+ public long getCommitMaxDelayMs() {
+ return getLong(COMMIT_MAX_DELAY_MS, DEFAULT_COMMIT_MAX_DELAY_MS);
+ }
+
+ public long getCommitTimeoutMs() {
+ return getLong(COMMIT_TIMEOUT_MS, DEFAULT_COMMIT_TIMEOUT_MS);
+ }
+
public Optional<String> getTaskClass() {
return Optional.ofNullable(get(TASK_CLASS));
}
@@ -315,6 +343,15 @@ public class TaskConfig extends MapConfig {
}
}
+ public List<Short> getCheckpointWriteVersions() {
+ return getList(CHECKPOINT_WRITE_VERSIONS,
DEFAULT_CHECKPOINT_WRITE_VERSIONS)
+ .stream().map(Short::valueOf).collect(Collectors.toList());
+ }
+
+ public short getCheckpointReadVersion() {
+ return getShort(CHECKPOINT_READ_VERSION, DEFAULT_CHECKPOINT_READ_VERSION);
+ }
+
public boolean getTransactionalStateCheckpointEnabled() {
return getBoolean(TRANSACTIONAL_STATE_CHECKPOINT_ENABLED,
DEFAULT_TRANSACTIONAL_STATE_CHECKPOINT_ENABLED);
}
diff --git
a/samza-core/src/main/java/org/apache/samza/serializers/CheckpointV2Serde.java
b/samza-core/src/main/java/org/apache/samza/serializers/CheckpointV2Serde.java
new file mode 100644
index 0000000..10e6b3d
--- /dev/null
+++
b/samza-core/src/main/java/org/apache/samza/serializers/CheckpointV2Serde.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.serializers;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.samza.SamzaException;
+import org.apache.samza.checkpoint.CheckpointId;
+import org.apache.samza.checkpoint.CheckpointV2;
+import org.apache.samza.serializers.model.SamzaObjectMapper;
+import org.apache.samza.system.SystemStreamPartition;
+
+
+/**
+ * The {@link Serde} for {@link CheckpointV2} which includes {@link
CheckpointId}s, state checkpoint markers
+ * and the input {@link SystemStreamPartition} offsets.
+ *
+ * The overall payload is serde'd as JSON using {@link SamzaObjectMapper}.
Since the Samza classes cannot be directly
+ * serialized by Jackson using {@link
org.apache.samza.serializers.model.JsonCheckpointV2Mixin}.
+ */
+public class CheckpointV2Serde implements Serde<CheckpointV2> {
+ private static final ObjectMapper OBJECT_MAPPER =
SamzaObjectMapper.getObjectMapper();
+
+ public CheckpointV2Serde() { }
+
+ @Override
+ public CheckpointV2 fromBytes(byte[] bytes) {
+ try {
+ return OBJECT_MAPPER.readValue(bytes, CheckpointV2.class);
+ } catch (Exception e) {
+ throw new SamzaException(String.format("Exception while deserializing
checkpoint: %s", new String(bytes)), e);
+ }
+ }
+
+ @Override
+ public byte[] toBytes(CheckpointV2 checkpoint) {
+ try {
+ return OBJECT_MAPPER.writeValueAsBytes(checkpoint);
+ } catch (Exception e) {
+ throw new SamzaException(String.format("Exception while serializing
checkpoint: %s", checkpoint.toString()), e);
+ }
+ }
+}
diff --git
a/samza-core/src/main/java/org/apache/samza/serializers/JsonCheckpoint.java
b/samza-core/src/main/java/org/apache/samza/serializers/JsonCheckpoint.java
new file mode 100644
index 0000000..47a8f73
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/serializers/JsonCheckpoint.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.serializers;
+
+import java.util.Map;
+
+/**
+ * Used for Json serialization of the {@link
org.apache.samza.checkpoint.Checkpoint} class by the
+ * {@link CheckpointV2Serde}
+ * This cannot be an internal class as required by Jackson Object mapper
+ */
+public class JsonCheckpoint {
+ private String checkpointId;
+ private Map<String, Map<String, String>> inputOffsets;
+ // Map<StorageBackendFactoryName, Map<StoreName, StateCheckpointMarker>>
+ private Map<String, Map<String, String>> stateCheckpointMarkers;
+
+ // Default constructor required for Jackson ObjectMapper
+ public JsonCheckpoint() {}
+
+ public JsonCheckpoint(String checkpointId,
+ Map<String, Map<String, String>> inputOffsets,
+ Map<String, Map<String, String>> stateCheckpointMakers) {
+ this.checkpointId = checkpointId;
+ this.inputOffsets = inputOffsets;
+ this.stateCheckpointMarkers = stateCheckpointMakers;
+ }
+
+ public String getCheckpointId() {
+ return checkpointId;
+ }
+
+ public Map<String, Map<String, String>> getInputOffsets() {
+ return inputOffsets;
+ }
+
+ public Map<String, Map<String, String>> getStateCheckpointMarkers() {
+ return stateCheckpointMarkers;
+ }
+}
diff --git
a/samza-core/src/main/java/org/apache/samza/serializers/model/JsonCheckpointV2Mixin.java
b/samza-core/src/main/java/org/apache/samza/serializers/model/JsonCheckpointV2Mixin.java
new file mode 100644
index 0000000..8e26745
--- /dev/null
+++
b/samza-core/src/main/java/org/apache/samza/serializers/model/JsonCheckpointV2Mixin.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.serializers.model;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.Map;
+import java.util.Set;
+import org.apache.samza.Partition;
+import org.apache.samza.checkpoint.CheckpointId;
+import org.apache.samza.container.TaskName;
+import org.apache.samza.system.SystemStreamPartition;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public abstract class JsonCheckpointV2Mixin {
+ @JsonCreator
+ public JsonCheckpointV2Mixin(
+ @JsonProperty("checkpoint-id") CheckpointId checkpointId,
+ @JsonProperty("input-offsets") Map<SystemStreamPartition, String>
inputOffsets,
+ @JsonProperty("state-checkpoint-markers") Map<String, Map<String,
String>> stateCheckpointMarkers) {
+ }
+
+ @JsonProperty("checkpoint-id")
+ abstract CheckpointId getCheckpointId();
+
+ @JsonProperty("input-offsets")
+ abstract Map<SystemStreamPartition, String> getOffsets();
+
+ @JsonProperty("state-checkpoint-markers")
+ abstract Map<String, Map<String, String>> getStateCheckpointMarkers();
+}
diff --git
a/samza-core/src/main/java/org/apache/samza/serializers/model/KafkaStateCheckpointMarkerMixin.java
b/samza-core/src/main/java/org/apache/samza/serializers/model/KafkaStateCheckpointMarkerMixin.java
new file mode 100644
index 0000000..6194877
--- /dev/null
+++
b/samza-core/src/main/java/org/apache/samza/serializers/model/KafkaStateCheckpointMarkerMixin.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.serializers.model;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.samza.system.SystemStreamPartition;
+
+
+/**
+ * A mix-in Jackson class to convert {@link
org.apache.samza.checkpoint.kafka.KafkaStateCheckpointMarker} to/from JSON
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+abstract public class KafkaStateCheckpointMarkerMixin {
+ @JsonCreator
+ public KafkaStateCheckpointMarkerMixin(
+ @JsonProperty("version") short version,
+ @JsonProperty("changelog-ssp") SystemStreamPartition changelogSSP,
+ @JsonProperty("changelog-offset") String changelogOffset) {
+ }
+
+ @JsonProperty("version")
+ abstract short getVersion();
+
+ @JsonProperty("changelog-ssp")
+ abstract SystemStreamPartition getChangelogSSP();
+
+ @JsonProperty("changelog-offset")
+ abstract String getChangelogOffset();
+}
diff --git
a/samza-core/src/main/java/org/apache/samza/serializers/model/SamzaObjectMapper.java
b/samza-core/src/main/java/org/apache/samza/serializers/model/SamzaObjectMapper.java
index 5c5078f..7888f1f 100644
---
a/samza-core/src/main/java/org/apache/samza/serializers/model/SamzaObjectMapper.java
+++
b/samza-core/src/main/java/org/apache/samza/serializers/model/SamzaObjectMapper.java
@@ -41,6 +41,9 @@ import
com.fasterxml.jackson.databind.introspect.AnnotatedMethod;
import com.fasterxml.jackson.databind.module.SimpleModule;
import org.apache.samza.Partition;
import org.apache.samza.SamzaException;
+import org.apache.samza.checkpoint.CheckpointId;
+import org.apache.samza.checkpoint.CheckpointV2;
+import org.apache.samza.checkpoint.kafka.KafkaStateCheckpointMarker;
import org.apache.samza.config.Config;
import org.apache.samza.config.MapConfig;
import org.apache.samza.container.TaskName;
@@ -95,11 +98,15 @@ public class SamzaObjectMapper {
module.addKeyDeserializer(SystemStreamPartition.class, new
SystemStreamPartitionKeyDeserializer());
module.addDeserializer(Config.class, new ConfigDeserializer());
module.addDeserializer(TaskMode.class, new TaskModeDeserializer());
+ module.addSerializer(CheckpointId.class, new CheckpointIdSerializer());
+ module.addDeserializer(CheckpointId.class, new CheckpointIdDeserializer());
// Setup mixins for data models.
mapper.addMixIn(TaskModel.class, JsonTaskModelMixIn.class);
mapper.addMixIn(ContainerModel.class, JsonContainerModelMixIn.class);
mapper.addMixIn(JobModel.class, JsonJobModelMixIn.class);
+ mapper.addMixIn(CheckpointV2.class, JsonCheckpointV2Mixin.class);
+ mapper.addMixIn(KafkaStateCheckpointMarker.class,
KafkaStateCheckpointMarkerMixin.class);
module.addDeserializer(ContainerModel.class, new
JsonDeserializer<ContainerModel>() {
@Override
@@ -255,6 +262,22 @@ public class SamzaObjectMapper {
}
}
+ public static class CheckpointIdSerializer extends
JsonSerializer<CheckpointId> {
+ @Override
+ public void serialize(CheckpointId checkpointId, JsonGenerator gen,
SerializerProvider serializers) throws IOException {
+ gen.writeString(checkpointId.serialize());
+ }
+ }
+
+ public static class CheckpointIdDeserializer extends
JsonDeserializer<CheckpointId> {
+ @Override
+ public CheckpointId deserialize(JsonParser jsonParser,
DeserializationContext context) throws IOException {
+ ObjectCodec oc = jsonParser.getCodec();
+ JsonNode node = oc.readTree(jsonParser);
+ return CheckpointId.deserialize(node.textValue());
+ }
+ }
+
/**
* A Jackson property naming strategy that converts camel case JSON fields to
* hyphenated names. For example, myVariableName would be converted to
diff --git
a/samza-core/src/main/scala/org/apache/samza/checkpoint/CheckpointTool.scala
b/samza-core/src/main/scala/org/apache/samza/checkpoint/CheckpointTool.scala
index 7936c0a..92b4e59 100644
--- a/samza-core/src/main/scala/org/apache/samza/checkpoint/CheckpointTool.scala
+++ b/samza-core/src/main/scala/org/apache/samza/checkpoint/CheckpointTool.scala
@@ -185,9 +185,10 @@ class CheckpointTool(newOffsets: TaskNameToCheckpointMap,
coordinatorStreamStore
taskNames.foreach(checkpointManager.register)
checkpointManager.start()
+ // TODO dchen make add support for checkpointv2
val lastCheckpoints = taskNames.map(taskName => {
taskName -> Option(checkpointManager.readLastCheckpoint(taskName))
- .getOrElse(new Checkpoint(new
java.util.HashMap[SystemStreamPartition, String]()))
+ .getOrElse(new CheckpointV1(new
java.util.HashMap[SystemStreamPartition, String]()))
.getOffsets
.asScala
.toMap
@@ -199,7 +200,7 @@ class CheckpointTool(newOffsets: TaskNameToCheckpointMap,
coordinatorStreamStore
newOffsets.foreach {
case (taskName: TaskName, offsets: Map[SystemStreamPartition,
String]) =>
logCheckpoint(taskName, offsets, "New offset to be written for
task: " + taskName)
- val checkpoint = new Checkpoint(offsets.asJava)
+ val checkpoint = new CheckpointV1(offsets.asJava)
checkpointManager.writeCheckpoint(taskName, checkpoint)
info(s"Updated the checkpoint of the task: $taskName to: $offsets")
}
diff --git
a/samza-core/src/main/scala/org/apache/samza/checkpoint/OffsetManager.scala
b/samza-core/src/main/scala/org/apache/samza/checkpoint/OffsetManager.scala
index 442d83f..7491eaa 100644
--- a/samza-core/src/main/scala/org/apache/samza/checkpoint/OffsetManager.scala
+++ b/samza-core/src/main/scala/org/apache/samza/checkpoint/OffsetManager.scala
@@ -19,6 +19,7 @@
package org.apache.samza.checkpoint
+import java.util
import java.util.HashMap
import java.util.concurrent.ConcurrentHashMap
@@ -220,6 +221,16 @@ class OffsetManager(
}
/**
+ * Get the last checkpoint saved in the checkpoint manager or null if there
is no recorded checkpoints for the task
+ */
+ def getLastTaskCheckpoint(taskName: TaskName): Checkpoint = {
+ if (checkpointManager != null) {
+ checkpointManager.readLastCheckpoint(taskName)
+ }
+ null
+ }
+
+ /**
* Get the starting offset for a SystemStreamPartition. This is the offset
* where a SamzaContainer begins reading from when it starts up.
*/
@@ -267,9 +278,9 @@ class OffsetManager(
* ensure there are no concurrent updates to the offsets between when this
method is
* invoked and the corresponding call to [[OffsetManager.writeCheckpoint()]]
*/
- def buildCheckpoint(taskName: TaskName): Checkpoint = {
+ def getLastProcessedOffsets(taskName: TaskName):
util.Map[SystemStreamPartition, String] = {
if (checkpointManager != null || checkpointListeners.nonEmpty) {
- debug("Getting checkpoint offsets for taskName %s." format taskName)
+ debug("Getting last processed offsets to checkpoint for taskName %s."
format taskName)
val taskStartingOffsets = startingOffsets.getOrElse(taskName,
throw new SamzaException("Couldn't find starting offsets for task: " +
taskName))
@@ -283,10 +294,10 @@ class OffsetManager(
.filterKeys(taskSSPs.contains)
val modifiedTaskOffsets = getModifiedOffsets(taskStartingOffsets,
taskLastProcessedOffsets)
- new Checkpoint(new HashMap(modifiedTaskOffsets)) // Copy into new Map to
prevent mutation
+ new util.HashMap(modifiedTaskOffsets)
} else {
- debug("Returning null checkpoint for taskName %s because no checkpoint
manager/callback is defined." format taskName)
- null
+ debug("Returning empty offsets for taskName %s because no checkpoint
manager/callback is defined." format taskName)
+ new util.HashMap()
}
}
@@ -336,11 +347,13 @@ class OffsetManager(
*/
def writeCheckpoint(taskName: TaskName, checkpoint: Checkpoint) {
if (checkpoint != null && (checkpointManager != null ||
checkpointListeners.nonEmpty)) {
- debug("Writing checkpoint for taskName %s with offsets %s." format
(taskName, checkpoint))
+ debug("Writing checkpoint for taskName: %s as: %s." format (taskName,
checkpoint))
+
+ val sspToOffsets = checkpoint.getOffsets
if(checkpointManager != null) {
checkpointManager.writeCheckpoint(taskName, checkpoint)
- val sspToOffsets = checkpoint.getOffsets
+
if(sspToOffsets != null) {
sspToOffsets.asScala.foreach {
case (ssp, cp) => {
@@ -357,7 +370,7 @@ class OffsetManager(
// changelog SSPs are not registered but may be present in the
Checkpoint if transactional state checkpointing
// is enabled.
val registeredSSPs = systemStreamPartitions.getOrElse(taskName,
Set[SystemStreamPartition]())
- checkpoint.getOffsets.asScala
+ sspToOffsets.asScala
.filterKeys(registeredSSPs.contains)
.groupBy { case (ssp, _) => ssp.getSystem }.foreach {
case (systemName:String, offsets: Map[SystemStreamPartition,
String]) => {
@@ -452,7 +465,6 @@ class OffsetManager(
Map(taskName -> checkpoint.getOffsets.asScala.toMap)
} else {
info("Did not receive a checkpoint for taskName %s. Proceeding without a
checkpoint." format taskName)
-
Map(taskName -> Map())
}
}
diff --git
a/samza-core/src/main/scala/org/apache/samza/checkpoint/file/FileSystemCheckpointManager.scala
b/samza-core/src/main/scala/org/apache/samza/checkpoint/file/FileSystemCheckpointManager.scala
index a9780b6..b740a13 100644
---
a/samza-core/src/main/scala/org/apache/samza/checkpoint/file/FileSystemCheckpointManager.scala
+++
b/samza-core/src/main/scala/org/apache/samza/checkpoint/file/FileSystemCheckpointManager.scala
@@ -24,13 +24,11 @@ import java.io.FileNotFoundException
import java.io.FileOutputStream
import org.apache.samza.SamzaException
-import org.apache.samza.checkpoint.Checkpoint
-import org.apache.samza.checkpoint.CheckpointManager
-import org.apache.samza.checkpoint.CheckpointManagerFactory
+import org.apache.samza.checkpoint.{Checkpoint, CheckpointManager,
CheckpointManagerFactory, CheckpointV1}
import org.apache.samza.config.{Config, FileSystemCheckpointManagerConfig,
JobConfig}
import org.apache.samza.container.TaskName
import org.apache.samza.metrics.MetricsRegistry
-import org.apache.samza.serializers.CheckpointSerde
+import org.apache.samza.serializers.CheckpointV1Serde
import org.apache.samza.util.ScalaJavaUtil.JavaOptionals
import scala.io.Source
@@ -38,14 +36,14 @@ import scala.io.Source
class FileSystemCheckpointManager(
jobName: String,
root: File,
- serde: CheckpointSerde = new
CheckpointSerde) extends CheckpointManager {
+ serde: CheckpointV1Serde = new
CheckpointV1Serde) extends CheckpointManager {
override def register(taskName: TaskName):Unit = Unit
def getCheckpointFile(taskName: TaskName) = getFile(jobName, taskName,
"checkpoints")
def writeCheckpoint(taskName: TaskName, checkpoint: Checkpoint) {
- val bytes = serde.toBytes(checkpoint)
+ val bytes = serde.toBytes(checkpoint.asInstanceOf[CheckpointV1])
val fos = new FileOutputStream(getCheckpointFile(taskName))
fos.write(bytes)
diff --git
a/samza-core/src/main/scala/org/apache/samza/serializers/CheckpointSerde.scala
b/samza-core/src/main/scala/org/apache/samza/serializers/CheckpointV1Serde.scala
similarity index 54%
rename from
samza-core/src/main/scala/org/apache/samza/serializers/CheckpointSerde.scala
rename to
samza-core/src/main/scala/org/apache/samza/serializers/CheckpointV1Serde.scala
index 0224667..7db709e 100644
---
a/samza-core/src/main/scala/org/apache/samza/serializers/CheckpointSerde.scala
+++
b/samza-core/src/main/scala/org/apache/samza/serializers/CheckpointV1Serde.scala
@@ -19,14 +19,13 @@
package org.apache.samza.serializers
-import com.fasterxml.jackson.core.`type`.TypeReference
import com.fasterxml.jackson.databind.ObjectMapper
-import org.apache.samza.util.Logging
import java.util
-import org.apache.samza.checkpoint.Checkpoint
-import org.apache.samza.container.TaskName
+import org.apache.samza.Partition
+import org.apache.samza.checkpoint.CheckpointV1
import org.apache.samza.system.SystemStreamPartition
-import org.apache.samza.{SamzaException, Partition}
+import org.apache.samza.util.Logging
+
import scala.collection.JavaConverters._
/**
@@ -35,34 +34,26 @@ import scala.collection.JavaConverters._
* then interfere with JSON's decoding of the overall map. We'll sidestep the
whole issue by turning the
* map into a list[String] of (System, Stream, Partition, Offset) serializing
that.
*/
-class CheckpointSerde extends Serde[Checkpoint] with Logging {
- import CheckpointSerde._
- // TODO: Elucidate the CheckpointSerde relationshiop to Serde. Should Serde
also have keyTo/FromBytes? Should
- // we just take CheckpointSerde here as interface and have this be
JSONCheckpointSerde?
- // TODO: Add more tests. This class currently only has direct test and is
mainly tested by the other checkpoint managers
+class CheckpointV1Serde extends Serde[CheckpointV1] with Logging {
val jsonMapper = new ObjectMapper()
- // Jackson absolutely hates Scala types and hidden conversions hate you, so
we're going to be very, very
- // explicit about the Java (not Scala) types used here and never let Scala
get its grubby little hands
- // on any instance.
-
- // Store checkpoint as maps keyed of the SSP.toString to the another map of
the constituent SSP components
- // and offset. Jackson can't automatically serialize the SSP since it's not
a POJO and this avoids
- // having to wrap it another class while maintaing readability.
-
- def fromBytes(bytes: Array[Byte]): Checkpoint = {
+ // Serialize checkpoint as maps keyed by the SSP.toString() to the another
map of the constituent SSP components
+ // and offset. Jackson can't automatically serialize the SSP since it's not
a POJO and this avoids
+ // having to wrap it another class while maintaining readability.
+ // { "SSP.toString()" -> {"system": system, "stream": stream, "partition":
partition, "offset": offset)}
+ def fromBytes(bytes: Array[Byte]): CheckpointV1 = {
try {
val jMap = jsonMapper.readValue(bytes, classOf[util.HashMap[String,
util.HashMap[String, String]]])
- def deserializeJSONMap(m:util.HashMap[String, String]) = {
- require(m.size() == 4, "All JSON-encoded SystemStreamPartitions must
have four keys")
- val system = m.get("system")
+ def deserializeJSONMap(sspInfo:util.HashMap[String, String]) = {
+ require(sspInfo.size() == 4, "All JSON-encoded SystemStreamPartitions
must have four keys")
+ val system = sspInfo.get("system")
require(system != null, "System must be present in JSON-encoded
SystemStreamPartition")
- val stream = m.get("stream")
+ val stream = sspInfo.get("stream")
require(stream != null, "Stream must be present in JSON-encoded
SystemStreamPartition")
- val partition = m.get("partition")
+ val partition = sspInfo.get("partition")
require(partition != null, "Partition must be present in JSON-encoded
SystemStreamPartition")
- val offset = m.get("offset")
+ val offset = sspInfo.get("offset")
// allow null offsets, e.g. for changelog ssps
new SystemStreamPartition(system, stream, new
Partition(partition.toInt)) -> offset
@@ -70,16 +61,15 @@ class CheckpointSerde extends Serde[Checkpoint] with
Logging {
val cpMap = jMap.values.asScala.map(deserializeJSONMap).toMap
- new Checkpoint(cpMap.asJava)
- }catch {
+ new CheckpointV1(cpMap.asJava)
+ } catch {
case e : Exception =>
- warn("Exception while deserializing checkpoint: " + e)
- debug("Exception detail:", e)
+ warn("Exception while deserializing checkpoint: {}",
util.Arrays.toString(bytes), e)
null
}
}
- def toBytes(checkpoint: Checkpoint): Array[Byte] = {
+ def toBytes(checkpoint: CheckpointV1): Array[Byte] = {
val offsets = checkpoint.getOffsets
val asMap = new util.HashMap[String, util.HashMap[String,
String]](offsets.size())
@@ -96,21 +86,4 @@ class CheckpointSerde extends Serde[Checkpoint] with Logging
{
jsonMapper.writeValueAsBytes(asMap)
}
-
- def changelogPartitionMappingFromBytes(bytes: Array[Byte]):
util.Map[TaskName, java.lang.Integer] = {
- try {
- jsonMapper.readValue(bytes, PARTITION_MAPPING_TYPEREFERENCE)
- } catch {
- case e : Exception =>
- throw new SamzaException("Exception while deserializing changelog
partition mapping", e)
- }
- }
-
- def changelogPartitionMappingToBytes(mapping: util.Map[TaskName,
java.lang.Integer]) = {
- jsonMapper.writeValueAsBytes(new util.HashMap[TaskName,
java.lang.Integer](mapping))
- }
-}
-
-object CheckpointSerde {
- val PARTITION_MAPPING_TYPEREFERENCE = new
TypeReference[util.HashMap[TaskName, java.lang.Integer]]() {}
-}
+}
\ No newline at end of file
diff --git a/samza-core/src/main/scala/org/apache/samza/util/FileUtil.scala
b/samza-core/src/main/scala/org/apache/samza/util/FileUtil.scala
index d416340..0298687 100644
--- a/samza-core/src/main/scala/org/apache/samza/util/FileUtil.scala
+++ b/samza-core/src/main/scala/org/apache/samza/util/FileUtil.scala
@@ -146,7 +146,14 @@ class FileUtil extends Logging {
}
def createDirectories(path: Path): Path = {
- Files.createDirectories(path)
+ // Files.createDirectories throws FileAlreadyExistsException if the path
already exists
+ // but the last dir in the path is a symlink to another dir. Check
explicitly if the path
+ // already exists to avoid this behavior.
+ if (!Files.exists(path)) {
+ Files.createDirectories(path)
+ } else {
+ path
+ }
}
/**
diff --git
a/samza-core/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaChangelogSSPOffset.java
b/samza-core/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaChangelogSSPOffset.java
new file mode 100644
index 0000000..324a349
--- /dev/null
+++
b/samza-core/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaChangelogSSPOffset.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.checkpoint.kafka;
+
+import org.apache.samza.checkpoint.CheckpointId;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+
+public class TestKafkaChangelogSSPOffset {
+ @Test
+ public void testSerializeDeserialize() {
+ KafkaChangelogSSPOffset kafkaChangelogSSPOffset = new
KafkaChangelogSSPOffset(CheckpointId.create(), "offset");
+ KafkaChangelogSSPOffset deserializedKafkaChangelogSSPOffset =
KafkaChangelogSSPOffset.fromString(kafkaChangelogSSPOffset.toString());
+
+ assertEquals(kafkaChangelogSSPOffset.getCheckpointId(),
deserializedKafkaChangelogSSPOffset.getCheckpointId());
+ assertEquals("offset",
deserializedKafkaChangelogSSPOffset.getChangelogOffset());
+ assertEquals(kafkaChangelogSSPOffset, deserializedKafkaChangelogSSPOffset);
+ }
+
+ @Test
+ public void testSerializeDeserializeNullOffsets() {
+ KafkaChangelogSSPOffset kafkaChangelogSSPOffset = new
KafkaChangelogSSPOffset(CheckpointId.create(), null);
+ KafkaChangelogSSPOffset deserializedKafkaChangelogSSPOffset =
KafkaChangelogSSPOffset.fromString(kafkaChangelogSSPOffset.toString());
+
+ assertEquals(kafkaChangelogSSPOffset.getCheckpointId(),
deserializedKafkaChangelogSSPOffset.getCheckpointId());
+ assertNull(deserializedKafkaChangelogSSPOffset.getChangelogOffset());
+ assertEquals(kafkaChangelogSSPOffset, deserializedKafkaChangelogSSPOffset);
+ }
+
+ @Test
+ public void testSerializationFormatForBackwardsCompatibility() {
+ KafkaChangelogSSPOffset kafkaChangelogSSPOffset = new
KafkaChangelogSSPOffset(CheckpointId.create(), "offset");
+
+ // WARNING: This format is written to persisted remotes stores and local
files, making a change in the format
+ // would be backwards incompatible
+ String expectedSerializationFormat =
kafkaChangelogSSPOffset.getCheckpointId() + KafkaChangelogSSPOffset.SEPARATOR +
+ kafkaChangelogSSPOffset.getChangelogOffset();
+ assertEquals(expectedSerializationFormat,
kafkaChangelogSSPOffset.toString());
+ assertEquals(kafkaChangelogSSPOffset,
KafkaChangelogSSPOffset.fromString(expectedSerializationFormat));
+ }
+
+ @Test
+ public void testNullSerializationFormatForBackwardsCompatibility() {
+ KafkaChangelogSSPOffset kafkaChangelogSSPOffset = new
KafkaChangelogSSPOffset(CheckpointId.create(), null);
+
+ // WARNING: This format is written to persisted remotes stores and local
files, making a change in the format
+ // would be backwards incompatible
+ String expectedSerializationFormat =
kafkaChangelogSSPOffset.getCheckpointId() + KafkaChangelogSSPOffset.SEPARATOR +
+ "null";
+ assertEquals(expectedSerializationFormat,
kafkaChangelogSSPOffset.toString());
+ assertEquals(kafkaChangelogSSPOffset,
KafkaChangelogSSPOffset.fromString(expectedSerializationFormat));
+ }
+}
diff --git
a/samza-core/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaStateCheckpointMarker.java
b/samza-core/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaStateCheckpointMarker.java
new file mode 100644
index 0000000..75eacff
--- /dev/null
+++
b/samza-core/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaStateCheckpointMarker.java
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.checkpoint.kafka;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.samza.Partition;
+import org.apache.samza.system.SystemStreamPartition;
+import org.junit.Test;
+import scala.Option;
+
+import static org.apache.samza.checkpoint.kafka.KafkaStateCheckpointMarker.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+
+public class TestKafkaStateCheckpointMarker {
+ @Test
+ public void testSerializeDeserialize() {
+ SystemStreamPartition ssp = new SystemStreamPartition("system", "stream",
new Partition(1));
+ KafkaStateCheckpointMarker marker = new KafkaStateCheckpointMarker(ssp,
"offset");
+ KafkaStateCheckpointMarker deserializedMarker = KafkaStateCheckpointMarker
+ .deserialize(KafkaStateCheckpointMarker.serialize(marker));
+
+ assertEquals(MARKER_VERSION, deserializedMarker.getVersion());
+ assertEquals(marker.getChangelogOffset(),
deserializedMarker.getChangelogOffset());
+ assertEquals(marker.getChangelogSSP(),
deserializedMarker.getChangelogSSP());
+ assertEquals(marker, deserializedMarker);
+ }
+
+ @Test
+ public void testSerializeDeserializeNullOffsets() {
+ SystemStreamPartition ssp = new SystemStreamPartition("system", "stream",
new Partition(1));
+ KafkaStateCheckpointMarker marker = new KafkaStateCheckpointMarker(ssp,
null);
+ KafkaStateCheckpointMarker deserializedMarker = KafkaStateCheckpointMarker
+ .deserialize(KafkaStateCheckpointMarker.serialize(marker));
+
+ assertEquals(MARKER_VERSION, deserializedMarker.getVersion());
+ assertNull(deserializedMarker.getChangelogOffset());
+ assertEquals(marker.getChangelogSSP(),
deserializedMarker.getChangelogSSP());
+ assertEquals(marker, deserializedMarker);
+ }
+
+ @Test
+ public void testStateCheckpointMarkerToSSPOffsetMap() {
+ SystemStreamPartition ssp1 = new SystemStreamPartition("system1",
"stream1", new Partition(1));
+ KafkaStateCheckpointMarker marker1 = new KafkaStateCheckpointMarker(ssp1,
"offset1");
+ SystemStreamPartition ssp2 = new SystemStreamPartition("system2",
"stream2", new Partition(2));
+ KafkaStateCheckpointMarker marker2 = new KafkaStateCheckpointMarker(ssp2,
null);
+ Map<String, String> storesToKSCM = ImmutableMap.of(
+ "store1", KafkaStateCheckpointMarker.serialize(marker1),
+ "store2", KafkaStateCheckpointMarker.serialize(marker2)
+ );
+ Map<String, Map<String, String>> factoryToSCMs = ImmutableMap.of(
+ KAFKA_STATE_BACKEND_FACTORY_NAME, storesToKSCM,
+ "factory2", Collections.EMPTY_MAP // factory2 should be ignored
+ );
+
+ Map<SystemStreamPartition, Option<String>> sspToOffsetOption =
KafkaStateCheckpointMarker
+ .scmsToSSPOffsetMap(factoryToSCMs);
+
+ assertEquals(2, sspToOffsetOption.size());
+ assertTrue(sspToOffsetOption.containsKey(ssp1));
+ assertEquals(sspToOffsetOption.get(ssp1).get(),
marker1.getChangelogOffset());
+ assertEquals(ssp1, marker1.getChangelogSSP());
+ assertTrue(sspToOffsetOption.containsKey(ssp2));
+ assertTrue(sspToOffsetOption.get(ssp2).isEmpty());
+ }
+
+ @Test
+ public void testStateCheckpointMarkerToSSPOffsetMapNoFactoryFound() {
+ Map<String, Map<String, String>> factoryToSCMs = ImmutableMap.of(
+ "factory1", Collections.EMPTY_MAP, // factory1 should be ignored
+ "factory2", Collections.EMPTY_MAP // factory2 should be ignored
+ );
+
+ Map<SystemStreamPartition, Option<String>> sspToOffsetOption =
KafkaStateCheckpointMarker
+ .scmsToSSPOffsetMap(factoryToSCMs);
+
+ assertEquals(0, sspToOffsetOption.size());
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testStateCheckpointMarkerToSSPOffsetMapDeserializationError() {
+ Map<String, String> storesToSCM = ImmutableMap.of(
+ "store1", "blobId-1234"
+ );
+ Map<String, Map<String, String>> factoryToSCMs = ImmutableMap.of(
+ "factory2", Collections.EMPTY_MAP, // factory2 should be ignored
+ KAFKA_STATE_BACKEND_FACTORY_NAME, storesToSCM
+ );
+
+ KafkaStateCheckpointMarker.scmsToSSPOffsetMap(factoryToSCMs);
+ }
+}
diff --git
a/samza-core/src/test/java/org/apache/samza/serializers/TestCheckpointV2Serde.java
b/samza-core/src/test/java/org/apache/samza/serializers/TestCheckpointV2Serde.java
new file mode 100644
index 0000000..d3b2d40
--- /dev/null
+++
b/samza-core/src/test/java/org/apache/samza/serializers/TestCheckpointV2Serde.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.serializers;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.samza.Partition;
+import org.apache.samza.checkpoint.CheckpointId;
+import org.apache.samza.checkpoint.CheckpointV2;
+import org.apache.samza.system.SystemStreamPartition;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+
+public class TestCheckpointV2Serde {
+
+ @Test
+ public void testCheckpointV2Serde() {
+ CheckpointV2Serde serde = new CheckpointV2Serde();
+ Map<SystemStreamPartition, String> offsets = new HashMap<>();
+ SystemStreamPartition systemStreamPartition = new
SystemStreamPartition("test-system", "test-stream", new Partition(777));
+ offsets.put(systemStreamPartition, "1");
+
+ // State Checkpoint marker
+ Map<String, Map<String, String>> factoryStateCheckpointMarkersMap = new
HashMap<>();
+ Map<String, String> stateCheckpointMarkersMap = new HashMap<>();
+ stateCheckpointMarkersMap.put("store1", "marker1");
+ stateCheckpointMarkersMap.put("store2", "marker2");
+
+ Map<String, String> stateCheckpointMarkersMap2 = new HashMap<>();
+ stateCheckpointMarkersMap2.put("store1", "marker3");
+ stateCheckpointMarkersMap2.put("store2", "marker4");
+
+ factoryStateCheckpointMarkersMap.put("factory1",
stateCheckpointMarkersMap);
+ factoryStateCheckpointMarkersMap.put("factory2",
stateCheckpointMarkersMap2);
+
+ CheckpointId checkpointId = CheckpointId.create();
+
+ CheckpointV2 checkpoint = new CheckpointV2(checkpointId, offsets,
factoryStateCheckpointMarkersMap);
+ CheckpointV2 deserializedCheckpoint =
serde.fromBytes(serde.toBytes(checkpoint));
+
+ // Validate input checkpoints
+ assertEquals(checkpointId, deserializedCheckpoint.getCheckpointId());
+ assertEquals("1",
deserializedCheckpoint.getOffsets().get(systemStreamPartition));
+ assertEquals(1, deserializedCheckpoint.getOffsets().size());
+
+ // Validate state checkpoints
+ assertEquals(2, deserializedCheckpoint.getStateCheckpointMarkers().size());
+
assertTrue(deserializedCheckpoint.getStateCheckpointMarkers().containsKey("factory1"));
+ assertEquals(stateCheckpointMarkersMap,
deserializedCheckpoint.getStateCheckpointMarkers().get("factory1"));
+
assertTrue(deserializedCheckpoint.getStateCheckpointMarkers().containsKey("factory2"));
+ assertEquals(stateCheckpointMarkersMap2,
deserializedCheckpoint.getStateCheckpointMarkers().get("factory2"));
+ }
+
+ @Test
+ public void testCheckpointV2SerdeStatelessJob() {
+ CheckpointV2Serde serde = new CheckpointV2Serde();
+ Map<SystemStreamPartition, String> offsets = new HashMap<>();
+ SystemStreamPartition systemStreamPartition = new
SystemStreamPartition("test-system", "test-stream", new Partition(777));
+ offsets.put(systemStreamPartition, "1");
+
+ // State Checkpoint marker
+ CheckpointId checkpointId = CheckpointId.create();
+
+
+ CheckpointV2 checkpoint = new CheckpointV2(checkpointId, offsets, new
HashMap<>());
+ CheckpointV2 deserializedCheckpoint =
serde.fromBytes(serde.toBytes(checkpoint));
+
+ // Validate input checkpoints
+ assertEquals(checkpointId, deserializedCheckpoint.getCheckpointId());
+ assertEquals("1",
deserializedCheckpoint.getOffsets().get(systemStreamPartition));
+ assertEquals(1, deserializedCheckpoint.getOffsets().size());
+
+ // No state checkpoints, but a map is still created
+ assertEquals(0, deserializedCheckpoint.getStateCheckpointMarkers().size());
+ }
+}
diff --git
a/samza-core/src/test/scala/org/apache/samza/checkpoint/TestCheckpointTool.scala
b/samza-core/src/test/scala/org/apache/samza/checkpoint/TestCheckpointTool.scala
index 777a6a0..5bcbdcd 100644
---
a/samza-core/src/test/scala/org/apache/samza/checkpoint/TestCheckpointTool.scala
+++
b/samza-core/src/test/scala/org/apache/samza/checkpoint/TestCheckpointTool.scala
@@ -91,9 +91,9 @@ class TestCheckpointTool extends AssertionsForJUnit with
MockitoSugar {
when(TestCheckpointTool.systemAdmin.getSystemStreamPartitionCounts(Set("foo").asJava,
0))
.thenReturn(Map("foo" -> metadata).asJava)
when(TestCheckpointTool.checkpointManager.readLastCheckpoint(tn0))
- .thenReturn(new Checkpoint(Map(new SystemStreamPartition("test", "foo",
p0) -> "1234").asJava))
+ .thenReturn(new CheckpointV1(Map(new SystemStreamPartition("test",
"foo", p0) -> "1234").asJava))
when(TestCheckpointTool.checkpointManager.readLastCheckpoint(tn1))
- .thenReturn(new Checkpoint(Map(new SystemStreamPartition("test", "foo",
p1) -> "4321").asJava))
+ .thenReturn(new CheckpointV1(Map(new SystemStreamPartition("test",
"foo", p1) -> "4321").asJava))
}
@Test
@@ -113,9 +113,9 @@ class TestCheckpointTool extends AssertionsForJUnit with
MockitoSugar {
val checkpointTool = CheckpointTool(config, toOverwrite)
checkpointTool.run()
verify(TestCheckpointTool.checkpointManager)
- .writeCheckpoint(tn0, new Checkpoint(Map(new
SystemStreamPartition("test", "foo", p0) -> "42").asJava))
+ .writeCheckpoint(tn0, new CheckpointV1(Map(new
SystemStreamPartition("test", "foo", p0) -> "42").asJava))
verify(TestCheckpointTool.checkpointManager)
- .writeCheckpoint(tn1, new Checkpoint(Map(new
SystemStreamPartition("test", "foo", p1) -> "43").asJava))
+ .writeCheckpoint(tn1, new CheckpointV1(Map(new
SystemStreamPartition("test", "foo", p1) -> "43").asJava))
}
@Test
@@ -157,9 +157,9 @@ class TestCheckpointTool extends AssertionsForJUnit with
MockitoSugar {
checkpointTool.run()
verify(TestCheckpointTool.checkpointManager)
- .writeCheckpoint(tn0, new Checkpoint(Map(new
SystemStreamPartition("test", "foo", p0) -> "42").asJava))
+ .writeCheckpoint(tn0, new CheckpointV1(Map(new
SystemStreamPartition("test", "foo", p0) -> "42").asJava))
verify(TestCheckpointTool.checkpointManager)
- .writeCheckpoint(tn1, new Checkpoint(Map(new
SystemStreamPartition("test", "foo", p1) -> "43").asJava))
+ .writeCheckpoint(tn1, new CheckpointV1(Map(new
SystemStreamPartition("test", "foo", p1) -> "43").asJava))
// Two configurations job.id, job.name are populated in the coordinator
config by SamzaRuntime and it is not present in generated config.
assert(generatedConfigs.entrySet().containsAll(TestCheckpointTool.coordinatorConfig.entrySet()))
diff --git
a/samza-core/src/test/scala/org/apache/samza/checkpoint/TestOffsetManager.scala
b/samza-core/src/test/scala/org/apache/samza/checkpoint/TestOffsetManager.scala
index 677504d..3949ecf 100644
---
a/samza-core/src/test/scala/org/apache/samza/checkpoint/TestOffsetManager.scala
+++
b/samza-core/src/test/scala/org/apache/samza/checkpoint/TestOffsetManager.scala
@@ -100,7 +100,7 @@ class TestOffsetManager {
}
startpointManagerUtil.getStartpointManager.start
assertFalse(startpointManagerUtil.getStartpointManager.getFanOutForTask(taskName).containsKey(systemStreamPartition))
// Startpoint should delete after checkpoint commit
- val expectedCheckpoint = new Checkpoint(Map(systemStreamPartition ->
"47").asJava)
+ val expectedCheckpoint = new CheckpointV1(Map(systemStreamPartition ->
"47").asJava)
assertEquals(expectedCheckpoint,
checkpointManager.readLastCheckpoint(taskName))
startpointManagerUtil.stop
}
@@ -260,7 +260,7 @@ class TestOffsetManager {
val systemStreamPartition = new SystemStreamPartition(systemStream,
partition)
val testStreamMetadata = new SystemStreamMetadata(systemStream.getStream,
Map(partition -> new SystemStreamPartitionMetadata("0", "1", "2")).asJava)
val systemStreamMetadata = Map(systemStream -> testStreamMetadata)
- val checkpoint = new Checkpoint(Map(systemStreamPartition -> "45").asJava)
+ val checkpoint = new CheckpointV1(Map(systemStreamPartition ->
"45").asJava)
val checkpointManager = getCheckpointManager(systemStreamPartition,
taskName)
val config = new MapConfig(Map(
"systems.test-system.samza.offset.default" -> "oldest",
@@ -289,7 +289,7 @@ class TestOffsetManager {
partition1 -> new SystemStreamPartitionMetadata("0", "1", "2"),
partition2 -> new SystemStreamPartitionMetadata("3", "4", "5")).asJava)
val systemStreamMetadata = Map(systemStream -> testStreamMetadata)
- val checkpoint = new Checkpoint(Map(systemStreamPartition1 -> "45").asJava)
+ val checkpoint = new CheckpointV1(Map(systemStreamPartition1 ->
"45").asJava)
// Checkpoint manager only has partition 1.
val checkpointManager = getCheckpointManager(systemStreamPartition1,
taskName1)
val startpointManagerUtil = getStartpointManagerUtil()
@@ -397,8 +397,7 @@ class TestOffsetManager {
val testStreamMetadata2 = new
SystemStreamMetadata(systemStream2.getStream, Map(partition -> new
SystemStreamPartitionMetadata("0", "1", "2")).asJava)
val systemStreamMetadata = Map(systemStream -> testStreamMetadata,
systemStream2->testStreamMetadata2)
val config = new MapConfig
- val checkpointManager = getCheckpointManager1(systemStreamPartition,
- new
Checkpoint(Map(systemStreamPartition -> "45", systemStreamPartition2 ->
"100").asJava),
+ val checkpointManager = getCheckpointManager1(new
CheckpointV1(Map(systemStreamPartition -> "45", systemStreamPartition2 ->
"100").asJava),
taskName)
val startpointManagerUtil = getStartpointManagerUtil()
val consumer = new SystemConsumerWithCheckpointCallback
@@ -423,9 +422,9 @@ class TestOffsetManager {
// Should get offset 45 back from the checkpoint manager, which is last
processed, and system admin should return 46 as starting offset.
assertTrue(startpointManagerUtil.getStartpointManager.getFanOutForTask(taskName).containsKey(systemStreamPartition))
val offsetsToCheckpoint = new java.util.HashMap[SystemStreamPartition,
String]()
-
offsetsToCheckpoint.putAll(offsetManager.buildCheckpoint(taskName).getOffsets)
+ offsetsToCheckpoint.putAll(offsetManager.getLastProcessedOffsets(taskName))
offsetsToCheckpoint.put(unregisteredSystemStreamPartition, "50")
- offsetManager.writeCheckpoint(taskName, new
Checkpoint(offsetsToCheckpoint))
+ offsetManager.writeCheckpoint(taskName, new
CheckpointV1(offsetsToCheckpoint))
intercept[IllegalStateException] {
// StartpointManager should stop after last fan out is removed
@@ -499,15 +498,15 @@ class TestOffsetManager {
offsetManager.update(taskName, systemStreamPartition, "46")
// Get checkpoint snapshot like we do at the beginning of
TaskInstance.commit()
- val checkpoint46 = offsetManager.buildCheckpoint(taskName)
+ val checkpoint46 = offsetManager.getLastProcessedOffsets(taskName)
offsetManager.update(taskName, systemStreamPartition, "47") // Offset
updated before checkpoint
- offsetManager.writeCheckpoint(taskName, checkpoint46)
+ offsetManager.writeCheckpoint(taskName, new CheckpointV1(checkpoint46))
assertEquals(Some("47"), offsetManager.getLastProcessedOffset(taskName,
systemStreamPartition))
assertEquals("46",
offsetManager.offsetManagerMetrics.checkpointedOffsets.get(systemStreamPartition).getValue)
// Now write the checkpoint for the latest offset
- val checkpoint47 = offsetManager.buildCheckpoint(taskName)
- offsetManager.writeCheckpoint(taskName, checkpoint47)
+ val checkpoint47 = offsetManager.getLastProcessedOffsets(taskName)
+ offsetManager.writeCheckpoint(taskName, new CheckpointV1(checkpoint47))
startpointManagerUtil.stop
assertEquals(Some("47"), offsetManager.getLastProcessedOffset(taskName,
systemStreamPartition))
assertEquals("47",
offsetManager.offsetManagerMetrics.checkpointedOffsets.get(systemStreamPartition).getValue)
@@ -582,7 +581,7 @@ class TestOffsetManager {
// Utility method to create and write checkpoint in one statement
def checkpoint(offsetManager: OffsetManager, taskName: TaskName): Unit = {
- offsetManager.writeCheckpoint(taskName,
offsetManager.buildCheckpoint(taskName))
+ offsetManager.writeCheckpoint(taskName, new
CheckpointV1(offsetManager.getLastProcessedOffsets(taskName)))
}
class SystemConsumerWithCheckpointCallback extends SystemConsumer with
CheckpointListener{
@@ -602,10 +601,10 @@ class TestOffsetManager {
}
private def getCheckpointManager(systemStreamPartition:
SystemStreamPartition, taskName:TaskName = new TaskName("taskName")) = {
- getCheckpointManager1(systemStreamPartition, new
Checkpoint(Map(systemStreamPartition -> "45").asJava), taskName)
+ getCheckpointManager1(new CheckpointV1(Map(systemStreamPartition ->
"45").asJava), taskName)
}
- private def getCheckpointManager1(systemStreamPartition:
SystemStreamPartition, checkpoint: Checkpoint, taskName:TaskName = new
TaskName("taskName")) = {
+ private def getCheckpointManager1(checkpoint: Checkpoint, taskName:TaskName
= new TaskName("taskName")) = {
new CheckpointManager {
var isStarted = false
var isStopped = false
@@ -617,9 +616,6 @@ class TestOffsetManager {
def writeCheckpoint(taskName: TaskName, checkpoint: Checkpoint) {
checkpoints += taskName -> checkpoint }
def readLastCheckpoint(taskName: TaskName) =
checkpoints.getOrElse(taskName, null)
def stop { isStopped = true }
-
- // Only for testing purposes - not present in actual checkpoint manager
- def getOffets = Map(taskName -> checkpoint.getOffsets.asScala.toMap)
}
}
diff --git
a/samza-core/src/test/scala/org/apache/samza/checkpoint/file/TestFileSystemCheckpointManager.scala
b/samza-core/src/test/scala/org/apache/samza/checkpoint/file/TestFileSystemCheckpointManager.scala
index f7839af..8cc6770 100644
---
a/samza-core/src/test/scala/org/apache/samza/checkpoint/file/TestFileSystemCheckpointManager.scala
+++
b/samza-core/src/test/scala/org/apache/samza/checkpoint/file/TestFileSystemCheckpointManager.scala
@@ -20,13 +20,15 @@
package org.apache.samza.checkpoint.file
import java.io.File
+
import scala.collection.JavaConverters._
import java.util.Random
+
import org.junit.Assert._
import org.junit.{After, Before, Test}
import org.apache.samza.SamzaException
import org.apache.samza.Partition
-import org.apache.samza.checkpoint.Checkpoint
+import org.apache.samza.checkpoint.{Checkpoint, CheckpointV1}
import org.apache.samza.system.SystemStreamPartition
import org.apache.samza.container.TaskName
import org.junit.rules.TemporaryFolder
@@ -52,17 +54,17 @@ class TestFileSystemCheckpointManager {
@Test
def testReadForCheckpointFileThatDoesExistShouldReturnProperCheckpoint {
- val cp = new Checkpoint(Map(
+ val cp = new CheckpointV1(Map(
new SystemStreamPartition("a", "b", new Partition(0)) -> "c",
new SystemStreamPartition("a", "c", new Partition(1)) -> "d",
new SystemStreamPartition("b", "d", new Partition(2)) -> "e").asJava)
- var readCp:Checkpoint = null
+ var readCp:CheckpointV1 = null
val cpm = new FileSystemCheckpointManager("some-job-name",
tempFolder.getRoot)
cpm.start
cpm.writeCheckpoint(taskName, cp)
- readCp = cpm.readLastCheckpoint(taskName)
+ readCp = cpm.readLastCheckpoint(taskName).asInstanceOf[CheckpointV1]
cpm.stop
assertNotNull(readCp)
diff --git
a/samza-core/src/test/scala/org/apache/samza/serializers/TestCheckpointSerde.scala
b/samza-core/src/test/scala/org/apache/samza/serializers/TestCheckpointV1Serde.scala
similarity index 69%
rename from
samza-core/src/test/scala/org/apache/samza/serializers/TestCheckpointSerde.scala
rename to
samza-core/src/test/scala/org/apache/samza/serializers/TestCheckpointV1Serde.scala
index c2060e0..02f2e59 100644
---
a/samza-core/src/test/scala/org/apache/samza/serializers/TestCheckpointSerde.scala
+++
b/samza-core/src/test/scala/org/apache/samza/serializers/TestCheckpointV1Serde.scala
@@ -22,46 +22,30 @@ package org.apache.samza.serializers
import java.util
import org.apache.samza.Partition
-import org.apache.samza.checkpoint.Checkpoint
+import org.apache.samza.checkpoint.{CheckpointV1}
import org.apache.samza.container.TaskName
import org.apache.samza.system.SystemStreamPartition
import org.junit.Assert._
import org.junit.Test
import scala.collection.JavaConverters._
-import scala.collection.mutable
-class TestCheckpointSerde {
+class TestCheckpointV1Serde {
@Test
def testExactlyOneOffset {
- val serde = new CheckpointSerde
+ val serde = new CheckpointV1Serde
var offsets = Map[SystemStreamPartition, String]()
val systemStreamPartition = new SystemStreamPartition("test-system",
"test-stream", new Partition(777))
offsets += systemStreamPartition -> "1"
- val deserializedOffsets = serde.fromBytes(serde.toBytes(new
Checkpoint(offsets.asJava)))
+ val deserializedOffsets = serde.fromBytes(serde.toBytes(new
CheckpointV1(offsets.asJava)))
assertEquals("1",
deserializedOffsets.getOffsets.get(systemStreamPartition))
assertEquals(1, deserializedOffsets.getOffsets.size)
}
@Test
- def testChangelogPartitionMappingRoundTrip {
- val mapping = new util.HashMap[TaskName, java.lang.Integer]()
- mapping.put(new TaskName("Ted"), 0)
- mapping.put(new TaskName("Dougal"), 1)
- mapping.put(new TaskName("Jack"), 2)
-
- val checkpointSerde = new CheckpointSerde
- val asBytes = checkpointSerde.changelogPartitionMappingToBytes(mapping)
- val backToMap = checkpointSerde.changelogPartitionMappingFromBytes(asBytes)
-
- assertEquals(mapping, backToMap)
- assertNotSame(mapping, backToMap)
- }
-
- @Test
def testNullCheckpointSerde: Unit = {
val checkpointBytes = null.asInstanceOf[Array[Byte]]
- val checkpointSerde = new CheckpointSerde
+ val checkpointSerde = new CheckpointV1Serde
val checkpoint = checkpointSerde.fromBytes(checkpointBytes)
assertNull(checkpoint)
}
diff --git a/samza-core/src/test/scala/org/apache/samza/util/TestFileUtil.scala
b/samza-core/src/test/scala/org/apache/samza/util/TestFileUtil.scala
index 6604ae0..104f983 100644
--- a/samza-core/src/test/scala/org/apache/samza/util/TestFileUtil.scala
+++ b/samza-core/src/test/scala/org/apache/samza/util/TestFileUtil.scala
@@ -21,11 +21,15 @@
package org.apache.samza.util
-import java.io.{File, FileInputStream, FileOutputStream, ObjectInputStream,
ObjectOutputStream}
+import org.apache.samza.testUtils.FileUtil
-import org.junit.Assert.{assertEquals, assertNull, assertTrue}
+import java.io.{File, FileInputStream, FileOutputStream, ObjectInputStream,
ObjectOutputStream}
+import org.junit.Assert.{assertEquals, assertNull, assertTrue, fail}
import org.junit.Test
+import java.nio.file.{FileAlreadyExistsException, Files, Paths}
+import scala.util.Random
+
class TestFileUtil {
val data = "100"
val fileUtil = new FileUtil()
@@ -104,4 +108,45 @@ class TestFileUtil {
// Check data returned
assertNull(result)
}
+
+ /**
+ * Files.createDirectories fails with a FileAlreadyExistsException if the
last directory
+ * in the path already exists but is a symlink to another directory. It
works correctly
+ * if one of the intermediate directory is a symlink. Verify this behavior
and
+ * test that the util method handles this correctly.
+ */
+ @Test
+ def testCreateDirectoriesWithSymlinks(): Unit = {
+ /**
+ * Directory structure:
+ * /tmp/samza-file-util-RANDOM
+ * /tmp/samza-file-util-RANDOM-symlink (symlink to dir above)
+ * /tmp/samza-file-util-RANDOM/subdir (created via the symlink above)
+ */
+ val tmpDirPath = Paths.get(FileUtil.TMP_DIR)
+ val tmpSubDirName = "samza-file-util-" + Random.nextInt()
+ val tmpSubDirSymlinkName = tmpSubDirName + "-symlink"
+
+ val tmpSubDirPath = Paths.get(FileUtil.TMP_DIR, tmpSubDirName);
+ fileUtil.createDirectories(tmpSubDirPath)
+
+ val tmpSymlinkPath = Paths.get(FileUtil.TMP_DIR, tmpSubDirSymlinkName)
+ Files.createSymbolicLink(tmpSymlinkPath, tmpDirPath);
+
+ try {
+ Files.createDirectories(tmpSymlinkPath)
+ fail("Should have thrown a FileAlreadyExistsException since last dir in
path already " +
+ "exists and is a symlink")
+ } catch {
+ case e: FileAlreadyExistsException =>
+ // ignore and continue
+ }
+
+ // test that the util method handles this correctly and does not throw an
exception
+ fileUtil.createDirectories(tmpSymlinkPath)
+
+ // verify that subdirs can be created via symlinks correctly.
+ val tmpSubSubDirPath = Paths.get(FileUtil.TMP_DIR, tmpSubDirName +
"-symlink", "subdir")
+ fileUtil.createDirectories(tmpSubSubDirPath)
+ }
}
diff --git
a/samza-kafka/src/main/java/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKey.java
b/samza-kafka/src/main/java/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKey.java
index 05114f9..0f0b071 100644
---
a/samza-kafka/src/main/java/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKey.java
+++
b/samza-kafka/src/main/java/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKey.java
@@ -26,7 +26,8 @@ import org.apache.samza.container.TaskName;
*/
public class KafkaCheckpointLogKey {
- public static final String CHECKPOINT_KEY_TYPE = "checkpoint";
+ public static final String CHECKPOINT_V1_KEY_TYPE = "checkpoint";
+ public static final String CHECKPOINT_V2_KEY_TYPE = "checkpoint-v2";
/**
* The SystemStreamPartitionGrouperFactory configured for this job run.
Since, checkpoints of different
* groupers are not compatible, we persist and validate them across job runs.
@@ -48,8 +49,9 @@ public class KafkaCheckpointLogKey {
Preconditions.checkNotNull(type);
Preconditions.checkState(!grouperFactoryClassName.isEmpty(), "Empty
grouper factory class provided");
- Preconditions.checkState(type.equals(CHECKPOINT_KEY_TYPE),
String.format("Invalid type provided for checkpoint key. " +
- "Expected: (%s) Actual: (%s)", CHECKPOINT_KEY_TYPE, type));
+ Preconditions.checkState(CHECKPOINT_V1_KEY_TYPE.equals(type) ||
CHECKPOINT_V2_KEY_TYPE.equals(type),
+ String.format("Invalid type provided for checkpoint key. Expected: (%s
or %s) Actual: (%s)",
+ CHECKPOINT_V1_KEY_TYPE, CHECKPOINT_V2_KEY_TYPE, type));
this.grouperFactoryClassName = grouperFactoryClassName;
this.taskName = taskName;
diff --git
a/samza-kafka/src/main/java/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKeySerde.java
b/samza-kafka/src/main/java/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKeySerde.java
index b00f12e..c5c431b 100644
---
a/samza-kafka/src/main/java/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKeySerde.java
+++
b/samza-kafka/src/main/java/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKeySerde.java
@@ -24,7 +24,6 @@ import org.apache.samza.SamzaException;
import org.apache.samza.container.TaskName;
import org.apache.samza.serializers.Serde;
-import java.util.Arrays;
import java.util.LinkedHashMap;
/**
@@ -58,16 +57,13 @@ public class KafkaCheckpointLogKeySerde implements
Serde<KafkaCheckpointLogKey>
public KafkaCheckpointLogKey fromBytes(byte[] bytes) {
try {
LinkedHashMap<String, String> deserializedKey = MAPPER.readValue(bytes,
LinkedHashMap.class);
+ String key = deserializedKey.get(TYPE_FIELD);
- if
(!KafkaCheckpointLogKey.CHECKPOINT_KEY_TYPE.equals(deserializedKey.get(TYPE_FIELD)))
{
- throw new IllegalArgumentException(String.format("Invalid key
detected. Type of the key is %s", deserializedKey.get(TYPE_FIELD)));
- }
-
- return new KafkaCheckpointLogKey(deserializedKey.get(TYPE_FIELD), new
TaskName(deserializedKey.get(TASK_NAME_FIELD)),
deserializedKey.get(SSP_GROUPER_FACTORY_FIELD)
- );
+ return new KafkaCheckpointLogKey(key, new
TaskName(deserializedKey.get(TASK_NAME_FIELD)),
+ deserializedKey.get(SSP_GROUPER_FACTORY_FIELD));
} catch (Exception e) {
throw new SamzaException(String.format("Exception in de-serializing
checkpoint bytes: %s",
- Arrays.toString(bytes)), e);
+ new String(bytes)), e);
}
}
}
diff --git
a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
b/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
index 757e7ae..a83a34b 100644
---
a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
+++
b/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
@@ -22,18 +22,16 @@ package org.apache.samza.checkpoint.kafka
import java.util.Collections
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicReference
-
import com.google.common.annotations.VisibleForTesting
import com.google.common.base.Preconditions
-import org.apache.samza.checkpoint.{Checkpoint, CheckpointManager}
+import org.apache.samza.checkpoint.{Checkpoint, CheckpointManager,
CheckpointV1, CheckpointV2}
import org.apache.samza.config.{Config, JobConfig, TaskConfig}
import org.apache.samza.container.TaskName
-import org.apache.samza.serializers.Serde
+import org.apache.samza.serializers.{CheckpointV1Serde, CheckpointV2Serde,
Serde}
import org.apache.samza.metrics.MetricsRegistry
-import org.apache.samza.serializers.CheckpointSerde
import org.apache.samza.system._
import org.apache.samza.system.kafka.KafkaStreamSpec
-import org.apache.samza.util.{ExponentialSleepStrategy, Logging}
+import org.apache.samza.util.Logging
import org.apache.samza.{Partition, SamzaException}
import scala.collection.mutable
@@ -53,7 +51,8 @@ class KafkaCheckpointManager(checkpointSpec: KafkaStreamSpec,
validateCheckpoint: Boolean,
config: Config,
metricsRegistry: MetricsRegistry,
- checkpointMsgSerde: Serde[Checkpoint] = new
CheckpointSerde,
+ checkpointV1MsgSerde: Serde[CheckpointV1] = new
CheckpointV1Serde,
+ checkpointV2MsgSerde: Serde[CheckpointV2] = new
CheckpointV2Serde,
checkpointKeySerde: Serde[KafkaCheckpointLogKey]
= new KafkaCheckpointLogKeySerde) extends CheckpointManager with Logging {
var MaxRetryDurationInMillis: Long = TimeUnit.MINUTES.toMillis(15)
@@ -81,6 +80,8 @@ class KafkaCheckpointManager(checkpointSpec: KafkaStreamSpec,
// for active containers, this will be set to true, while false for standby
containers.
val stopConsumerAfterFirstRead: Boolean = new
TaskConfig(config).getCheckpointManagerConsumerStopAfterFirstRead
+ val checkpointReadVersion: Short = new
TaskConfig(config).getCheckpointReadVersion
+
/**
* Create checkpoint stream prior to start.
*
@@ -159,19 +160,7 @@ class KafkaCheckpointManager(checkpointSpec:
KafkaStreamSpec,
* @inheritdoc
*/
override def writeCheckpoint(taskName: TaskName, checkpoint: Checkpoint) {
- val key = new
KafkaCheckpointLogKey(KafkaCheckpointLogKey.CHECKPOINT_KEY_TYPE, taskName,
expectedGrouperFactory)
- val keyBytes = try {
- checkpointKeySerde.toBytes(key)
- } catch {
- case e: Exception => throw new SamzaException(s"Exception when writing
checkpoint-key for $taskName: $checkpoint", e)
- }
- val msgBytes = try {
- checkpointMsgSerde.toBytes(checkpoint)
- } catch {
- case e: Exception => throw new SamzaException(s"Exception when writing
checkpoint for $taskName: $checkpoint", e)
- }
-
- val envelope = new OutgoingMessageEnvelope(checkpointSsp, keyBytes,
msgBytes)
+ val envelope = buildOutgoingMessageEnvelope(taskName, checkpoint)
// Used for exponential backoff retries on failure in sending messages
through producer.
val startTimeInMillis: Long = System.currentTimeMillis()
@@ -188,7 +177,7 @@ class KafkaCheckpointManager(checkpointSpec:
KafkaStreamSpec,
} catch {
case exception: Exception => {
producerException = exception
- warn(s"Retrying failed checkpoint write to key: $key, checkpoint:
$checkpoint for task: $taskName", exception)
+ warn(s"Retrying failed write for checkpoint: $checkpoint for task:
$taskName", exception)
// TODO: Remove this producer recreation logic after SAMZA-1393.
val newProducer: SystemProducer = getSystemProducer()
producerCreationLock.synchronized {
@@ -276,10 +265,10 @@ class KafkaCheckpointManager(checkpointSpec:
KafkaStreamSpec,
checkpointKeySerde.fromBytes(keyBytes)
} catch {
case e: Exception => if (validateCheckpoint) {
- throw new SamzaException(s"Exception while serializing
checkpoint-key. " +
+ throw new SamzaException(s"Exception while deserializing
checkpoint-key. " +
s"Topic: $checkpointTopic Offset: $offset", e)
} else {
- warn(s"Ignoring exception while serializing checkpoint-key. Topic:
$checkpointTopic Offset: $offset", e)
+ warn(s"Ignoring exception while deserializing checkpoint-key. Topic:
$checkpointTopic Offset: $offset", e)
null
}
}
@@ -295,17 +284,27 @@ class KafkaCheckpointManager(checkpointSpec:
KafkaStreamSpec,
}
}
- // If the type of the key is not
KafkaCheckpointLogKey.CHECKPOINT_KEY_TYPE, it can safely be ignored.
- if
(KafkaCheckpointLogKey.CHECKPOINT_KEY_TYPE.equals(checkpointKey.getType)) {
- val checkpointBytes =
checkpointEnvelope.getMessage.asInstanceOf[Array[Byte]]
- val checkpoint = try {
- checkpointMsgSerde.fromBytes(checkpointBytes)
- } catch {
- case e: Exception => throw new SamzaException(s"Exception while
serializing checkpoint-message. " +
- s"Topic: $checkpointTopic Offset: $offset", e)
- }
-
- checkpoints.put(checkpointKey.getTaskName, checkpoint)
+ val msgBytes = checkpointEnvelope.getMessage.asInstanceOf[Array[Byte]]
+ try {
+ // if checkpoint key version does not match configured checkpoint
version to read, skip the message.
+ if (checkpointReadVersion == CheckpointV1.CHECKPOINT_VERSION &&
+
KafkaCheckpointLogKey.CHECKPOINT_V1_KEY_TYPE.equals(checkpointKey.getType)) {
+ val msgBytes =
checkpointEnvelope.getMessage.asInstanceOf[Array[Byte]]
+ val checkpoint = checkpointV1MsgSerde.fromBytes(msgBytes)
+ checkpoints.put(checkpointKey.getTaskName, checkpoint)
+ } else if (checkpointReadVersion == CheckpointV2.CHECKPOINT_VERSION
&&
+
KafkaCheckpointLogKey.CHECKPOINT_V2_KEY_TYPE.equals(checkpointKey.getType)) {
+ val checkpoint = checkpointV2MsgSerde.fromBytes(msgBytes)
+ checkpoints.put(checkpointKey.getTaskName, checkpoint)
+ } // else ignore and skip the message
+ } catch {
+ case e: Exception =>
+ if (validateCheckpoint) {
+ throw new SamzaException(s"Exception while deserializing
checkpoint-message. " +
+ s"Topic: $checkpointTopic Offset: $offset", e)
+ } else {
+ warn(s"Ignoring exception while deserializing checkpoint-msg.
Topic: $checkpointTopic Offset: $offset", e)
+ }
}
}
}
@@ -333,4 +332,44 @@ class KafkaCheckpointManager(checkpointSpec:
KafkaStreamSpec,
partitionMetaData.getOldestOffset
}
+
+ private def buildOutgoingMessageEnvelope[T <: Checkpoint](taskName:
TaskName, checkpoint: T): OutgoingMessageEnvelope = {
+ checkpoint match {
+ case checkpointV1: CheckpointV1 => {
+ val key = new KafkaCheckpointLogKey(
+ KafkaCheckpointLogKey.CHECKPOINT_V1_KEY_TYPE, taskName,
expectedGrouperFactory)
+ val keyBytes = try {
+ checkpointKeySerde.toBytes(key)
+ } catch {
+ case e: Exception =>
+ throw new SamzaException(s"Exception when writing checkpoint-key
for $taskName: $checkpoint", e)
+ }
+ val msgBytes = try {
+ checkpointV1MsgSerde.toBytes(checkpointV1)
+ } catch {
+ case e: Exception =>
+ throw new SamzaException(s"Exception when writing checkpoint for
$taskName: $checkpoint", e)
+ }
+ new OutgoingMessageEnvelope(checkpointSsp, keyBytes, msgBytes)
+ }
+ case checkpointV2: CheckpointV2 => {
+ val key = new KafkaCheckpointLogKey(
+ KafkaCheckpointLogKey.CHECKPOINT_V2_KEY_TYPE, taskName,
expectedGrouperFactory)
+ val keyBytes = try {
+ checkpointKeySerde.toBytes(key)
+ } catch {
+ case e: Exception =>
+ throw new SamzaException(s"Exception when writing checkpoint-key
for $taskName: $checkpoint", e)
+ }
+ val msgBytes = try {
+ checkpointV2MsgSerde.toBytes(checkpointV2)
+ } catch {
+ case e: Exception =>
+ throw new SamzaException(s"Exception when writing checkpoint for
$taskName: $checkpoint", e)
+ }
+ new OutgoingMessageEnvelope(checkpointSsp, keyBytes, msgBytes)
+ }
+ case _ => throw new SamzaException("Unknown checkpoint version: " +
checkpoint.getVersion)
+ }
+ }
}
diff --git
a/samza-kafka/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointLogKeySerde.java
b/samza-kafka/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointLogKeySerde.java
index b648b1c..7245e70 100644
---
a/samza-kafka/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointLogKeySerde.java
+++
b/samza-kafka/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointLogKeySerde.java
@@ -29,7 +29,7 @@ public class TestKafkaCheckpointLogKeySerde {
@Test
public void testBinaryCompatibility() {
- KafkaCheckpointLogKey logKey1 = new
KafkaCheckpointLogKey(KafkaCheckpointLogKey.CHECKPOINT_KEY_TYPE,
+ KafkaCheckpointLogKey logKey1 = new
KafkaCheckpointLogKey(KafkaCheckpointLogKey.CHECKPOINT_V1_KEY_TYPE,
new TaskName("Partition 0"),
GroupByPartitionFactory.class.getCanonicalName());
KafkaCheckpointLogKeySerde checkpointSerde = new
KafkaCheckpointLogKeySerde();
@@ -43,11 +43,21 @@ public class TestKafkaCheckpointLogKeySerde {
@Test
public void testSerde() {
- KafkaCheckpointLogKey key = new
KafkaCheckpointLogKey(KafkaCheckpointLogKey.CHECKPOINT_KEY_TYPE,
+ KafkaCheckpointLogKey key = new
KafkaCheckpointLogKey(KafkaCheckpointLogKey.CHECKPOINT_V1_KEY_TYPE,
new TaskName("Partition 0"),
GroupByPartitionFactory.class.getCanonicalName());
KafkaCheckpointLogKeySerde checkpointSerde = new
KafkaCheckpointLogKeySerde();
// test that deserialize(serialize(k)) == k
Assert.assertEquals(key,
checkpointSerde.fromBytes(checkpointSerde.toBytes(key)));
}
+
+ @Test
+ public void testCheckpointTypeV2() {
+ KafkaCheckpointLogKey keyV2 = new
KafkaCheckpointLogKey(KafkaCheckpointLogKey.CHECKPOINT_V2_KEY_TYPE,
+ new TaskName("Partition 0"),
GroupByPartitionFactory.class.getCanonicalName());
+ KafkaCheckpointLogKeySerde checkpointKeySerde = new
KafkaCheckpointLogKeySerde();
+
+ // test that deserialize(serialize(k)) == k
+ Assert.assertEquals(keyV2,
checkpointKeySerde.fromBytes(checkpointKeySerde.toBytes(keyV2)));
+ }
}
diff --git
a/samza-kafka/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManagerJava.java
b/samza-kafka/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManagerJava.java
index f32040a..d0e927f 100644
---
a/samza-kafka/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManagerJava.java
+++
b/samza-kafka/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManagerJava.java
@@ -24,13 +24,14 @@ import kafka.common.KafkaException;
import kafka.common.TopicAlreadyMarkedForDeletionException;
import org.apache.samza.Partition;
import org.apache.samza.SamzaException;
-import org.apache.samza.checkpoint.Checkpoint;
+import org.apache.samza.checkpoint.CheckpointV1;
import org.apache.samza.config.Config;
import org.apache.samza.config.JobConfig;
import org.apache.samza.container.TaskName;
import org.apache.samza.container.grouper.stream.GroupByPartitionFactory;
import org.apache.samza.metrics.MetricsRegistry;
-import org.apache.samza.serializers.CheckpointSerde;
+import org.apache.samza.serializers.CheckpointV1Serde;
+import org.apache.samza.serializers.CheckpointV2Serde;
import org.apache.samza.system.IncomingMessageEnvelope;
import org.apache.samza.system.StreamValidationException;
import org.apache.samza.system.SystemAdmin;
@@ -72,7 +73,7 @@ public class TestKafkaCheckpointManagerJava {
SystemFactory factory = newFactory(mock(SystemProducer.class),
mock(SystemConsumer.class), mockAdmin);
KafkaCheckpointManager checkpointManager = new
KafkaCheckpointManager(checkpointSpec, factory,
- true, mock(Config.class), mock(MetricsRegistry.class), null, new
KafkaCheckpointLogKeySerde());
+ true, mock(Config.class), mock(MetricsRegistry.class), null, null, new
KafkaCheckpointLogKeySerde());
// expect an exception during startup
checkpointManager.createResources();
@@ -91,7 +92,7 @@ public class TestKafkaCheckpointManagerJava {
SystemFactory factory = newFactory(mock(SystemProducer.class),
mock(SystemConsumer.class), mockAdmin);
KafkaCheckpointManager checkpointManager = new
KafkaCheckpointManager(checkpointSpec, factory,
- true, mock(Config.class), mock(MetricsRegistry.class), null, new
KafkaCheckpointLogKeySerde());
+ true, mock(Config.class), mock(MetricsRegistry.class), null, null, new
KafkaCheckpointLogKeySerde());
// expect an exception during startup
checkpointManager.createResources();
@@ -116,7 +117,7 @@ public class TestKafkaCheckpointManagerJava {
// wire up an exception throwing serde with the checkpointmanager
KafkaCheckpointManager checkpointManager = new
KafkaCheckpointManager(checkpointSpec, factory,
- true, mockConfig, mock(MetricsRegistry.class), new
ExceptionThrowingCheckpointSerde(), new KafkaCheckpointLogKeySerde());
+ true, mockConfig, mock(MetricsRegistry.class), new
ExceptionThrowingCheckpointV1Serde(), null, new KafkaCheckpointLogKeySerde());
checkpointManager.register(TASK1);
checkpointManager.start();
@@ -142,7 +143,7 @@ public class TestKafkaCheckpointManagerJava {
// wire up an exception throwing serde with the checkpointmanager
KafkaCheckpointManager checkpointManager = new
KafkaCheckpointManager(checkpointSpec, factory,
- false, mockConfig, mock(MetricsRegistry.class), new
ExceptionThrowingCheckpointSerde(),
+ false, mockConfig, mock(MetricsRegistry.class), new
ExceptionThrowingCheckpointV1Serde(), null,
new ExceptionThrowingCheckpointKeySerde());
checkpointManager.register(TASK1);
checkpointManager.start();
@@ -167,7 +168,8 @@ public class TestKafkaCheckpointManagerJava {
SystemAdmin mockAdmin = newAdmin(oldestOffset, "1");
SystemFactory factory = newFactory(mock(SystemProducer.class),
mockConsumer, mockAdmin);
KafkaCheckpointManager checkpointManager = new
KafkaCheckpointManager(checkpointSpec, factory,
- true, mockConfig, mock(MetricsRegistry.class), new CheckpointSerde(),
new KafkaCheckpointLogKeySerde());
+ true, mockConfig, mock(MetricsRegistry.class), new
CheckpointV1Serde(), new CheckpointV2Serde(),
+ new KafkaCheckpointLogKeySerde());
checkpointManager.register(TASK1);
// 1. verify that consumer.register is called only during
checkpointManager.start.
@@ -206,12 +208,13 @@ public class TestKafkaCheckpointManagerJava {
SystemFactory factory = newFactory(mock(SystemProducer.class),
mockConsumer, mockAdmin);
KafkaCheckpointManager checkpointManager = new
KafkaCheckpointManager(checkpointSpec, factory,
- true, mockConfig, mock(MetricsRegistry.class), new CheckpointSerde(),
new KafkaCheckpointLogKeySerde());
+ true, mockConfig, mock(MetricsRegistry.class), new
CheckpointV1Serde(), new CheckpointV2Serde(),
+ new KafkaCheckpointLogKeySerde());
checkpointManager.register(TASK1);
checkpointManager.start();
// check that all ten messages are read, and the checkpoint is the newest
message
- Checkpoint checkpoint = checkpointManager.readLastCheckpoint(TASK1);
+ CheckpointV1 checkpoint = (CheckpointV1)
checkpointManager.readLastCheckpoint(TASK1);
Assert.assertEquals(checkpoint.getOffsets(), ImmutableMap.of(ssp,
Integer.toString(newestOffset)));
}
@@ -261,15 +264,15 @@ public class TestKafkaCheckpointManagerJava {
new KafkaCheckpointLogKey("checkpoint", taskName,
GROUPER_FACTORY_CLASS);
KafkaCheckpointLogKeySerde checkpointKeySerde = new
KafkaCheckpointLogKeySerde();
- Checkpoint checkpointMsg = new Checkpoint(ImmutableMap.of(ssp, offset));
- CheckpointSerde checkpointMsgSerde = new CheckpointSerde();
+ CheckpointV1 checkpointMsg = new CheckpointV1(ImmutableMap.of(ssp,
offset));
+ CheckpointV1Serde checkpointMsgSerde = new CheckpointV1Serde();
return new IncomingMessageEnvelope(CHECKPOINT_SSP, offset,
checkpointKeySerde.toBytes(checkpointKey),
checkpointMsgSerde.toBytes(checkpointMsg));
}
- private static class ExceptionThrowingCheckpointSerde extends
CheckpointSerde {
- public Checkpoint fromBytes(byte[] bytes) {
+ private static class ExceptionThrowingCheckpointV1Serde extends
CheckpointV1Serde {
+ public CheckpointV1 fromBytes(byte[] bytes) {
throw new KafkaException("exception");
}
}
diff --git
a/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
b/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
index 7d6db64..f940292 100644
---
a/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
+++
b/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
@@ -24,12 +24,12 @@ import java.util.Properties
import kafka.integration.KafkaServerTestHarness
import kafka.utils.{CoreUtils, TestUtils}
import com.google.common.collect.ImmutableMap
-import org.apache.samza.checkpoint.Checkpoint
+import org.apache.samza.checkpoint.{Checkpoint, CheckpointId, CheckpointV1,
CheckpointV2}
import org.apache.samza.config._
import org.apache.samza.container.TaskName
import org.apache.samza.container.grouper.stream.GroupByPartitionFactory
import org.apache.samza.metrics.MetricsRegistry
-import org.apache.samza.serializers.CheckpointSerde
+import org.apache.samza.serializers.CheckpointV1Serde
import org.apache.samza.system._
import org.apache.samza.system.kafka.{KafkaStreamSpec, KafkaSystemFactory}
import org.apache.samza.util.ScalaJavaUtil.JavaOptionals
@@ -48,8 +48,8 @@ class TestKafkaCheckpointManager extends
KafkaServerTestHarness {
val sspGrouperFactoryName = classOf[GroupByPartitionFactory].getCanonicalName
val ssp = new SystemStreamPartition("kafka", "topic", new Partition(0))
- val checkpoint1 = new Checkpoint(ImmutableMap.of(ssp, "offset-1"))
- val checkpoint2 = new Checkpoint(ImmutableMap.of(ssp, "offset-2"))
+ val checkpoint1 = new CheckpointV1(ImmutableMap.of(ssp, "offset-1"))
+ val checkpoint2 = new CheckpointV1(ImmutableMap.of(ssp, "offset-2"))
val taskName = new TaskName("Partition 0")
var config: Config = null
@@ -88,7 +88,7 @@ class TestKafkaCheckpointManager extends
KafkaServerTestHarness {
checkPointManager.register(taskName)
checkPointManager.start
- checkPointManager.writeCheckpoint(taskName, new
Checkpoint(ImmutableMap.of()))
+ checkPointManager.writeCheckpoint(taskName, new
CheckpointV1(ImmutableMap.of()))
checkPointManager.stop()
// Verifications after the test
@@ -127,6 +127,102 @@ class TestKafkaCheckpointManager extends
KafkaServerTestHarness {
}
@Test
+ def testCheckpointV1AndV2WriteAndReadV1(): Unit = {
+ val checkpointTopic = "checkpoint-topic-1"
+ val kcm1 = createKafkaCheckpointManager(checkpointTopic)
+ kcm1.register(taskName)
+ kcm1.createResources
+ kcm1.start
+ kcm1.stop
+
+ // check that start actually creates the topic with log compaction enabled
+ val topicConfig =
adminZkClient.getAllTopicConfigs().getOrElse(checkpointTopic, new Properties())
+
+ assertEquals(topicConfig, new
KafkaConfig(config).getCheckpointTopicProperties())
+ assertEquals("compact", topicConfig.get("cleanup.policy"))
+ assertEquals("26214400", topicConfig.get("segment.bytes"))
+
+ // read before topic exists should result in a null checkpoint
+ val readCp = readCheckpoint(checkpointTopic, taskName)
+ assertNull(readCp)
+
+ val checkpointV1 = new CheckpointV1(ImmutableMap.of(ssp, "offset-1"))
+ val checkpointV2 = new CheckpointV2(CheckpointId.create(),
ImmutableMap.of(ssp, "offset-2"),
+ ImmutableMap.of("factory1", ImmutableMap.of("store1",
"changelogOffset")))
+
+ // skips v2 checkpoints from checkpoint topic
+ writeCheckpoint(checkpointTopic, taskName, checkpointV2)
+ assertNull(readCheckpoint(checkpointTopic, taskName))
+
+ // reads latest v1 checkpoints
+ writeCheckpoint(checkpointTopic, taskName, checkpointV1)
+ assertEquals(checkpointV1, readCheckpoint(checkpointTopic, taskName))
+
+ // writing checkpoint v2 still returns the previous v1 checkpoint
+ writeCheckpoint(checkpointTopic, taskName, checkpointV2)
+ assertEquals(checkpointV1, readCheckpoint(checkpointTopic, taskName))
+ }
+
+ @Test
+ def testCheckpointV1AndV2WriteAndReadV2(): Unit = {
+ val checkpointTopic = "checkpoint-topic-1"
+ val kcm1 = createKafkaCheckpointManager(checkpointTopic)
+ kcm1.register(taskName)
+ kcm1.createResources
+ kcm1.start
+ kcm1.stop
+
+ // check that start actually creates the topic with log compaction enabled
+ val topicConfig =
adminZkClient.getAllTopicConfigs().getOrElse(checkpointTopic, new Properties())
+
+ assertEquals(topicConfig, new
KafkaConfig(config).getCheckpointTopicProperties())
+ assertEquals("compact", topicConfig.get("cleanup.policy"))
+ assertEquals("26214400", topicConfig.get("segment.bytes"))
+
+ // read before topic exists should result in a null checkpoint
+ val readCp = readCheckpoint(checkpointTopic, taskName)
+ assertNull(readCp)
+
+ val checkpointV1 = new CheckpointV1(ImmutableMap.of(ssp, "offset-1"))
+ val checkpointV2 = new CheckpointV2(CheckpointId.create(),
ImmutableMap.of(ssp, "offset-2"),
+ ImmutableMap.of("factory1", ImmutableMap.of("store1",
"changelogOffset")))
+
+ val overrideConfig = new MapConfig(new ImmutableMap.Builder[String,
String]()
+ .put(JobConfig.JOB_NAME, "some-job-name")
+ .put(JobConfig.JOB_ID, "i001")
+ .put(s"systems.$checkpointSystemName.samza.factory",
classOf[KafkaSystemFactory].getCanonicalName)
+ .put(s"systems.$checkpointSystemName.producer.bootstrap.servers",
brokerList)
+ .put(s"systems.$checkpointSystemName.consumer.zookeeper.connect",
zkConnect)
+ .put("task.checkpoint.system", checkpointSystemName)
+ .put(TaskConfig.CHECKPOINT_READ_VERSION, "2")
+ .build())
+
+ // Skips reading any v1 checkpoints
+ writeCheckpoint(checkpointTopic, taskName, checkpointV1)
+ assertNull(readCheckpoint(checkpointTopic, taskName, overrideConfig))
+
+ // writing a v2 checkpoint would allow reading it back
+ writeCheckpoint(checkpointTopic, taskName, checkpointV2)
+ assertEquals(checkpointV2, readCheckpoint(checkpointTopic, taskName,
overrideConfig))
+
+ // writing v1 checkpoint is still skipped
+ writeCheckpoint(checkpointTopic, taskName, checkpointV1)
+ assertEquals(checkpointV2, readCheckpoint(checkpointTopic, taskName,
overrideConfig))
+ }
+
+ @Test
+ def testCheckpointValidationSkipped(): Unit = {
+ val checkpointTopic = "checkpoint-topic-1"
+ val kcm1 = createKafkaCheckpointManager(checkpointTopic, serde = new
MockCheckpointSerde(),
+ failOnTopicValidation = false)
+ kcm1.register(taskName)
+ kcm1.start
+ kcm1.writeCheckpoint(taskName, new CheckpointV1(ImmutableMap.of(ssp,
"offset-1")))
+ kcm1.readLastCheckpoint(taskName)
+ kcm1.stop
+ }
+
+ @Test
def testWriteCheckpointShouldRetryFiniteTimesOnFailure(): Unit = {
val checkpointTopic = "checkpoint-topic-2"
val mockKafkaProducer: SystemProducer =
Mockito.mock(classOf[SystemProducer])
@@ -142,7 +238,7 @@ class TestKafkaCheckpointManager extends
KafkaServerTestHarness {
try {
checkPointManager.register(taskName)
checkPointManager.start
- checkPointManager.writeCheckpoint(taskName, new
Checkpoint(ImmutableMap.of()))
+ checkPointManager.writeCheckpoint(taskName, new
CheckpointV1(ImmutableMap.of()))
} catch {
case _: SamzaException => info("Got SamzaException as expected.")
case unexpectedException: Throwable => fail("Expected SamzaException but
got %s" format unexpectedException)
@@ -175,7 +271,7 @@ class TestKafkaCheckpointManager extends
KafkaServerTestHarness {
// create topic with the wrong number of partitions
createTopic(checkpointTopic, 8, new
KafkaConfig(config).getCheckpointTopicProperties())
val failOnTopicValidation = false
- val kcm = createKafkaCheckpointManager(checkpointTopic, new
CheckpointSerde, failOnTopicValidation)
+ val kcm = createKafkaCheckpointManager(checkpointTopic, new
CheckpointV1Serde, failOnTopicValidation)
kcm.register(taskName)
kcm.createResources()
kcm.start()
@@ -251,24 +347,25 @@ class TestKafkaCheckpointManager extends
KafkaServerTestHarness {
.build())
}
- private def createKafkaCheckpointManager(cpTopic: String, serde:
CheckpointSerde = new CheckpointSerde, failOnTopicValidation: Boolean = true) =
{
- val kafkaConfig = new org.apache.samza.config.KafkaConfig(config)
+ private def createKafkaCheckpointManager(cpTopic: String, serde:
CheckpointV1Serde = new CheckpointV1Serde,
+ failOnTopicValidation: Boolean = true, overrideConfig: Config = config) = {
+ val kafkaConfig = new org.apache.samza.config.KafkaConfig(overrideConfig)
val props = kafkaConfig.getCheckpointTopicProperties()
val systemName = kafkaConfig.getCheckpointSystem.getOrElse(
throw new SamzaException("No system defined for Kafka's checkpoint
manager."))
- val systemConfig = new SystemConfig(config)
+ val systemConfig = new SystemConfig(overrideConfig)
val systemFactoryClassName =
JavaOptionals.toRichOptional(systemConfig.getSystemFactory(systemName)).toOption
.getOrElse(throw new SamzaException("Missing configuration: " +
SystemConfig.SYSTEM_FACTORY_FORMAT format systemName))
val systemFactory = ReflectionUtil.getObj(systemFactoryClassName,
classOf[SystemFactory])
val spec = new KafkaStreamSpec("id", cpTopic, checkpointSystemName, 1, 1,
props)
- new KafkaCheckpointManager(spec, systemFactory, failOnTopicValidation,
config, new NoOpMetricsRegistry, serde)
+ new KafkaCheckpointManager(spec, systemFactory, failOnTopicValidation,
overrideConfig, new NoOpMetricsRegistry, serde)
}
- private def readCheckpoint(checkpointTopic: String, taskName: TaskName) :
Checkpoint = {
- val kcm = createKafkaCheckpointManager(checkpointTopic)
+ private def readCheckpoint(checkpointTopic: String, taskName: TaskName,
config: Config = config) : Checkpoint = {
+ val kcm = createKafkaCheckpointManager(checkpointTopic, overrideConfig =
config)
kcm.register(taskName)
kcm.start
val checkpoint = kcm.readLastCheckpoint(taskName)
@@ -300,4 +397,10 @@ class TestKafkaCheckpointManager extends
KafkaServerTestHarness {
}
}
+ class MockCheckpointSerde() extends CheckpointV1Serde {
+ override def fromBytes(bytes: Array[Byte]): CheckpointV1 = {
+ throw new SamzaException("Failed to deserialize")
+ }
+ }
+
}
diff --git
a/samza-test/src/test/java/org/apache/samza/checkpoint/CheckpointVersionIntegrationTest.java
b/samza-test/src/test/java/org/apache/samza/checkpoint/CheckpointVersionIntegrationTest.java
new file mode 100644
index 0000000..d0f1c2f
--- /dev/null
+++
b/samza-test/src/test/java/org/apache/samza/checkpoint/CheckpointVersionIntegrationTest.java
@@ -0,0 +1,149 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.checkpoint;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.samza.config.JobConfig;
+import org.apache.samza.config.JobCoordinatorConfig;
+import org.apache.samza.config.KafkaConfig;
+import org.apache.samza.config.TaskConfig;
+import org.apache.samza.storage.MyStatefulApplication;
+import org.apache.samza.test.framework.StreamApplicationIntegrationTestHarness;
+import org.apache.samza.util.FileUtil;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class CheckpointVersionIntegrationTest extends
StreamApplicationIntegrationTestHarness {
+
+ private final static Logger LOG =
LoggerFactory.getLogger(CheckpointVersionIntegrationTest.class);
+
+ private static final String INPUT_TOPIC = "inputTopic";
+ private static final String INPUT_SYSTEM = "kafka";
+ private static final String STORE_NAME = "store";
+ private static final String CHANGELOG_TOPIC = "changelog";
+ private static final String LOGGED_STORE_BASE_DIR = new
File(System.getProperty("java.io.tmpdir"), "logged-store").getAbsolutePath();
+ private static final Map<String, String> CONFIGS = new HashMap<String,
String>() { {
+ put(JobCoordinatorConfig.JOB_COORDINATOR_FACTORY,
"org.apache.samza.standalone.PassthroughJobCoordinatorFactory");
+ put(JobConfig.PROCESSOR_ID, "0");
+ put(TaskConfig.GROUPER_FACTORY,
"org.apache.samza.container.grouper.task.GroupByContainerIdsFactory");
+ put(TaskConfig.CHECKPOINT_MANAGER_FACTORY,
"org.apache.samza.checkpoint.kafka.KafkaCheckpointManagerFactory");
+ put(TaskConfig.COMMIT_MS, "-1"); // manual commit only
+ put(TaskConfig.TRANSACTIONAL_STATE_RESTORE_ENABLED, "true");
+ put(TaskConfig.TRANSACTIONAL_STATE_RETAIN_EXISTING_STATE, "true");
+ put(KafkaConfig.CHECKPOINT_REPLICATION_FACTOR(), "1");
+ put(JobConfig.JOB_LOGGED_STORE_BASE_DIR, LOGGED_STORE_BASE_DIR);
+ put(TaskConfig.COMMIT_MAX_DELAY_MS, "0"); // Ensure no commits are
skipped due to in progress commits
+ } };
+
+ @Before
+ @Override
+ public void setUp() {
+ super.setUp();
+ // reset static state shared with task between each parameterized iteration
+ MyStatefulApplication.resetTestState();
+ new FileUtil().rm(new File(LOGGED_STORE_BASE_DIR)); // always clear local
store on startup
+ }
+
+ @Test
+ public void testStopCheckpointV1V2AndRestartCheckpointV2() {
+ List<String> inputMessagesOnInitialRun = Arrays.asList("1", "2", "3", "2",
"97", "-97", ":98", ":99", ":crash_once");
+ // double check collectors.flush
+ List<String> expectedChangelogMessagesOnInitialRun = Arrays.asList("1",
"2", "3", "2", "97", null, "98", "99");
+ initialRun(inputMessagesOnInitialRun,
expectedChangelogMessagesOnInitialRun);
+
+ // first two are reverts for uncommitted messages from last run for keys
98 and 99
+ List<String> expectedChangelogMessagesAfterSecondRun =
+ Arrays.asList(null, null, "98", "99", "4", "5", "5");
+ List<String> expectedInitialStoreContentsOnSecondRun = Arrays.asList("1",
"2", "3");
+ Map<String, String> configOverrides = new HashMap<>(CONFIGS);
+ configOverrides.put(TaskConfig.CHECKPOINT_READ_VERSION, "2");
+ secondRun(CHANGELOG_TOPIC,
+ expectedChangelogMessagesAfterSecondRun,
expectedInitialStoreContentsOnSecondRun, configOverrides);
+ }
+
+ private void initialRun(List<String> inputMessages, List<String>
expectedChangelogMessages) {
+ // create input topic and produce the first batch of input messages
+ createTopic(INPUT_TOPIC, 1);
+ inputMessages.forEach(m -> produceMessage(INPUT_TOPIC, 0, m, m));
+
+ // verify that the input messages were produced successfully
+ if (inputMessages.size() > 0) {
+ List<ConsumerRecord<String, String>> inputRecords =
+ consumeMessages(Collections.singletonList(INPUT_TOPIC),
inputMessages.size());
+ List<String> readInputMessages =
inputRecords.stream().map(ConsumerRecord::value).collect(Collectors.toList());
+ Assert.assertEquals(inputMessages, readInputMessages);
+ }
+
+ // run the application
+ RunApplicationContext context = runApplication(
+ new MyStatefulApplication(INPUT_SYSTEM, INPUT_TOPIC,
Collections.singletonMap(STORE_NAME, CHANGELOG_TOPIC)), "myApp", CONFIGS);
+
+ // wait for the application to finish
+ context.getRunner().waitForFinish();
+
+ // consume and verify the changelog messages
+ if (expectedChangelogMessages.size() > 0) {
+ List<ConsumerRecord<String, String>> changelogRecords =
+ consumeMessages(Collections.singletonList(CHANGELOG_TOPIC),
expectedChangelogMessages.size());
+ List<String> changelogMessages =
changelogRecords.stream().map(ConsumerRecord::value).collect(Collectors.toList());
+ Assert.assertEquals(expectedChangelogMessages, changelogMessages);
+ }
+
+ LOG.info("Finished initial run");
+ }
+
+ private void secondRun(String changelogTopic, List<String>
expectedChangelogMessages,
+ List<String> expectedInitialStoreContents, Map<String, String>
overriddenConfigs) {
+ // remove previous files so restore is from the checkpointV2
+ new FileUtil().rm(new File(LOGGED_STORE_BASE_DIR));
+
+ // produce the second batch of input messages
+
+ List<String> inputMessages = Arrays.asList("4", "5", "5", ":shutdown");
+ inputMessages.forEach(m -> produceMessage(INPUT_TOPIC, 0, m, m));
+
+ // run the application
+ RunApplicationContext context = runApplication(
+ new MyStatefulApplication(INPUT_SYSTEM, INPUT_TOPIC,
Collections.singletonMap(STORE_NAME, changelogTopic)), "myApp",
overriddenConfigs);
+
+ // wait for the application to finish
+ context.getRunner().waitForFinish();
+
+ // verify the store contents during startup (this is after changelog
verification to ensure init has completed)
+ Assert.assertEquals(expectedInitialStoreContents,
MyStatefulApplication.getInitialStoreContents().get(STORE_NAME));
+
+ // consume and verify any additional changelog messages
+ List<ConsumerRecord<String, String>> changelogRecords =
+ consumeMessages(Collections.singletonList(changelogTopic),
expectedChangelogMessages.size());
+ List<String> changelogMessages =
changelogRecords.stream().map(ConsumerRecord::value).collect(Collectors.toList());
+ Assert.assertEquals(expectedChangelogMessages, changelogMessages);
+ }
+}
diff --git
a/samza-test/src/test/java/org/apache/samza/storage/kv/TransactionalStateIntegrationTest.java
b/samza-test/src/test/java/org/apache/samza/storage/kv/TransactionalStateIntegrationTest.java
index 05e6737..e9b12a1 100644
---
a/samza-test/src/test/java/org/apache/samza/storage/kv/TransactionalStateIntegrationTest.java
+++
b/samza-test/src/test/java/org/apache/samza/storage/kv/TransactionalStateIntegrationTest.java
@@ -92,6 +92,7 @@ public class TransactionalStateIntegrationTest extends
StreamApplicationIntegrat
put(TaskConfig.TRANSACTIONAL_STATE_RETAIN_EXISTING_STATE, "true");
put(KafkaConfig.CHECKPOINT_REPLICATION_FACTOR(), "1");
put(JobConfig.JOB_LOGGED_STORE_BASE_DIR, LOGGED_STORE_BASE_DIR);
+ put(TaskConfig.COMMIT_MAX_DELAY_MS, "0"); // Ensure no commits are
skipped due to in progress commits
} };
private static List<String> actualInitialStoreContents = new ArrayList<>();
diff --git
a/samza-test/src/test/java/org/apache/samza/storage/kv/TransactionalStateMultiStoreIntegrationTest.java
b/samza-test/src/test/java/org/apache/samza/storage/kv/TransactionalStateMultiStoreIntegrationTest.java
index 41eb1ab..d50d6bf 100644
---
a/samza-test/src/test/java/org/apache/samza/storage/kv/TransactionalStateMultiStoreIntegrationTest.java
+++
b/samza-test/src/test/java/org/apache/samza/storage/kv/TransactionalStateMultiStoreIntegrationTest.java
@@ -91,6 +91,7 @@ public class TransactionalStateMultiStoreIntegrationTest
extends StreamApplicati
put(TaskConfig.TRANSACTIONAL_STATE_RETAIN_EXISTING_STATE, "true");
put(KafkaConfig.CHECKPOINT_REPLICATION_FACTOR(), "1");
put(JobConfig.JOB_LOGGED_STORE_BASE_DIR, LOGGED_STORE_BASE_DIR);
+ put(TaskConfig.COMMIT_MAX_DELAY_MS, "0"); // Ensure no commits are
skipped due to in progress commits
} };
private static List<String> actualInitialStoreContents = new ArrayList<>();
