You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by fa...@apache.org on 2026/03/23 15:06:38 UTC
(pekko) branch main updated: [AKKA-31844] dynamic scaling of ShardedDaemonProcess instances (#2755)
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 4e8f81908c [AKKA-31844] dynamic scaling of ShardedDaemonProcess instances (#2755)
4e8f81908c is described below
commit 4e8f81908c0faad2be80924e1b898e6181b5d0df
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Mon Mar 23 16:06:33 2026 +0100
[AKKA-31844] dynamic scaling of ShardedDaemonProcess instances (#2755)
* Initial plan
* Changes before error encountered
Co-authored-by: pjfanning <11...@users.noreply.github.com>
* Port Akka PR 31844: dynamic scaling of ShardedDaemonProcess
Co-authored-by: pjfanning <11...@users.noreply.github.com>
* scalafmt
* Update ReplicatorMessages.java
* update ShardingMessages
* Complete test updates for ShardedDaemonProcess rescaling port
Co-authored-by: pjfanning <11...@users.noreply.github.com>
* Remove unused StopShardTimeout class to fix build warning
Co-authored-by: pjfanning <11...@users.noreply.github.com>
* Create dynamic-scaling.excludes
* Update dynamic-scaling.excludes
---------
Co-authored-by: copilot-swe-agent[bot] <19...@users.noreply.github.com>
Co-authored-by: pjfanning <11...@users.noreply.github.com>
---
.../typed/internal/protobuf/ShardingMessages.java | 2847 +++++++++++++++++++-
.../dynamic-scaling.excludes | 21 +
.../src/main/protobuf/ShardingMessages.proto | 25 +-
.../src/main/resources/reference.conf | 2 +-
.../typed/ShardedDaemonProcessCommand.scala | 119 +
.../typed/ShardedDaemonProcessSettings.scala | 22 +
.../sharding/typed/ShardingMessageExtractor.scala | 5 +-
.../ClusterShardingTypedSerializable.scala | 22 +
.../internal/ShardedDaemonProcessCoordinator.scala | 395 +++
.../typed/internal/ShardedDaemonProcessId.scala | 76 +
.../typed/internal/ShardedDaemonProcessImpl.scala | 278 +-
.../typed/internal/ShardedDaemonProcessState.scala | 148 +
.../typed/internal/ShardingSerializer.scala | 148 +-
.../typed/javadsl/ShardedDaemonProcess.scala | 58 +-
.../typed/scaladsl/ShardedDaemonProcess.scala | 55 +
.../ShardedDaemonProcessCompileOnlyTest.java | 61 +
.../internal/ShardedDaemonProcessIdSpec.scala | 67 +
.../internal/ShardedDaemonProcessStateSpec.scala | 45 +
.../typed/scaladsl/ShardedDaemonProcessSpec.scala | 60 +-
.../pekko/cluster/sharding/ShardCoordinator.scala | 28 +-
.../pekko/cluster/sharding/ShardRegion.scala | 5 +
.../ddata/protobuf/msg/ReplicatorMessages.java | 83 +-
22 files changed, 4389 insertions(+), 181 deletions(-)
diff --git a/cluster-sharding-typed/src/main/java/org/apache/pekko/cluster/sharding/typed/internal/protobuf/ShardingMessages.java b/cluster-sharding-typed/src/main/java/org/apache/pekko/cluster/sharding/typed/internal/protobuf/ShardingMessages.java
index e07b9fe515..c13d152f36 100644
--- a/cluster-sharding-typed/src/main/java/org/apache/pekko/cluster/sharding/typed/internal/protobuf/ShardingMessages.java
+++ b/cluster-sharding-typed/src/main/java/org/apache/pekko/cluster/sharding/typed/internal/protobuf/ShardingMessages.java
@@ -807,11 +807,2819 @@ public final class ShardingMessages extends org.apache.pekko.protobufv3.internal
}
+ public interface DaemonProcessScaleStateOrBuilder extends
+ // @@protoc_insertion_point(interface_extends:org.apache.pekko.cluster.sharding.typed.DaemonProcessScaleState)
+ org.apache.pekko.protobufv3.internal.MessageOrBuilder {
+
+ /**
+ * <code>required int64 revision = 1;</code>
+ * @return Whether the revision field is set.
+ */
+ boolean hasRevision();
+ /**
+ * <code>required int64 revision = 1;</code>
+ * @return The revision.
+ */
+ long getRevision();
+
+ /**
+ * <code>required int32 numberOfProcesses = 2;</code>
+ * @return Whether the numberOfProcesses field is set.
+ */
+ boolean hasNumberOfProcesses();
+ /**
+ * <code>required int32 numberOfProcesses = 2;</code>
+ * @return The numberOfProcesses.
+ */
+ int getNumberOfProcesses();
+
+ /**
+ * <code>required bool completed = 3;</code>
+ * @return Whether the completed field is set.
+ */
+ boolean hasCompleted();
+ /**
+ * <code>required bool completed = 3;</code>
+ * @return The completed.
+ */
+ boolean getCompleted();
+
+ /**
+ * <code>required int64 startedTimestampMillis = 4;</code>
+ * @return Whether the startedTimestampMillis field is set.
+ */
+ boolean hasStartedTimestampMillis();
+ /**
+ * <code>required int64 startedTimestampMillis = 4;</code>
+ * @return The startedTimestampMillis.
+ */
+ long getStartedTimestampMillis();
+ }
+ /**
+ * Protobuf type {@code org.apache.pekko.cluster.sharding.typed.DaemonProcessScaleState}
+ */
+ public static final class DaemonProcessScaleState extends
+ org.apache.pekko.protobufv3.internal.GeneratedMessage implements
+ // @@protoc_insertion_point(message_implements:org.apache.pekko.cluster.sharding.typed.DaemonProcessScaleState)
+ DaemonProcessScaleStateOrBuilder {
+ private static final long serialVersionUID = 0L;
+ static {
+ org.apache.pekko.protobufv3.internal.RuntimeVersion.validateProtobufGencodeVersion(
+ org.apache.pekko.protobufv3.internal.RuntimeVersion.RuntimeDomain.PUBLIC,
+ /* major= */ 4,
+ /* minor= */ 33,
+ /* patch= */ 0,
+ /* suffix= */ "",
+ "DaemonProcessScaleState");
+ }
+ // Use DaemonProcessScaleState.newBuilder() to construct.
+ private DaemonProcessScaleState(org.apache.pekko.protobufv3.internal.GeneratedMessage.Builder<?> builder) {
+ super(builder);
+ }
+ private DaemonProcessScaleState() {
+ }
+
+ public static final org.apache.pekko.protobufv3.internal.Descriptors.Descriptor
+ getDescriptor() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_DaemonProcessScaleState_descriptor;
+ }
+
+ @java.lang.Override
+ protected org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable
+ internalGetFieldAccessorTable() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_DaemonProcessScaleState_fieldAccessorTable
+ .ensureFieldAccessorsInitialized(
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState.class, org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState.Builder.class);
+ }
+
+ private int bitField0_;
+ public static final int REVISION_FIELD_NUMBER = 1;
+ private long revision_ = 0L;
+ /**
+ * <code>required int64 revision = 1;</code>
+ * @return Whether the revision field is set.
+ */
+ @java.lang.Override
+ public boolean hasRevision() {
+ return ((bitField0_ & 0x00000001) != 0);
+ }
+ /**
+ * <code>required int64 revision = 1;</code>
+ * @return The revision.
+ */
+ @java.lang.Override
+ public long getRevision() {
+ return revision_;
+ }
+
+ public static final int NUMBEROFPROCESSES_FIELD_NUMBER = 2;
+ private int numberOfProcesses_ = 0;
+ /**
+ * <code>required int32 numberOfProcesses = 2;</code>
+ * @return Whether the numberOfProcesses field is set.
+ */
+ @java.lang.Override
+ public boolean hasNumberOfProcesses() {
+ return ((bitField0_ & 0x00000002) != 0);
+ }
+ /**
+ * <code>required int32 numberOfProcesses = 2;</code>
+ * @return The numberOfProcesses.
+ */
+ @java.lang.Override
+ public int getNumberOfProcesses() {
+ return numberOfProcesses_;
+ }
+
+ public static final int COMPLETED_FIELD_NUMBER = 3;
+ private boolean completed_ = false;
+ /**
+ * <code>required bool completed = 3;</code>
+ * @return Whether the completed field is set.
+ */
+ @java.lang.Override
+ public boolean hasCompleted() {
+ return ((bitField0_ & 0x00000004) != 0);
+ }
+ /**
+ * <code>required bool completed = 3;</code>
+ * @return The completed.
+ */
+ @java.lang.Override
+ public boolean getCompleted() {
+ return completed_;
+ }
+
+ public static final int STARTEDTIMESTAMPMILLIS_FIELD_NUMBER = 4;
+ private long startedTimestampMillis_ = 0L;
+ /**
+ * <code>required int64 startedTimestampMillis = 4;</code>
+ * @return Whether the startedTimestampMillis field is set.
+ */
+ @java.lang.Override
+ public boolean hasStartedTimestampMillis() {
+ return ((bitField0_ & 0x00000008) != 0);
+ }
+ /**
+ * <code>required int64 startedTimestampMillis = 4;</code>
+ * @return The startedTimestampMillis.
+ */
+ @java.lang.Override
+ public long getStartedTimestampMillis() {
+ return startedTimestampMillis_;
+ }
+
+ private byte memoizedIsInitialized = -1;
+ @java.lang.Override
+ public final boolean isInitialized() {
+ byte isInitialized = memoizedIsInitialized;
+ if (isInitialized == 1) return true;
+ if (isInitialized == 0) return false;
+
+ if (!hasRevision()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
+ if (!hasNumberOfProcesses()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
+ if (!hasCompleted()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
+ if (!hasStartedTimestampMillis()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
+ memoizedIsInitialized = 1;
+ return true;
+ }
+
+ @java.lang.Override
+ public void writeTo(org.apache.pekko.protobufv3.internal.CodedOutputStream output)
+ throws java.io.IOException {
+ if (((bitField0_ & 0x00000001) != 0)) {
+ output.writeInt64(1, revision_);
+ }
+ if (((bitField0_ & 0x00000002) != 0)) {
+ output.writeInt32(2, numberOfProcesses_);
+ }
+ if (((bitField0_ & 0x00000004) != 0)) {
+ output.writeBool(3, completed_);
+ }
+ if (((bitField0_ & 0x00000008) != 0)) {
+ output.writeInt64(4, startedTimestampMillis_);
+ }
+ getUnknownFields().writeTo(output);
+ }
+
+ @java.lang.Override
+ public int getSerializedSize() {
+ int size = memoizedSize;
+ if (size != -1) return size;
+
+ size = 0;
+ if (((bitField0_ & 0x00000001) != 0)) {
+ size += org.apache.pekko.protobufv3.internal.CodedOutputStream
+ .computeInt64Size(1, revision_);
+ }
+ if (((bitField0_ & 0x00000002) != 0)) {
+ size += org.apache.pekko.protobufv3.internal.CodedOutputStream
+ .computeInt32Size(2, numberOfProcesses_);
+ }
+ if (((bitField0_ & 0x00000004) != 0)) {
+ size += org.apache.pekko.protobufv3.internal.CodedOutputStream
+ .computeBoolSize(3, completed_);
+ }
+ if (((bitField0_ & 0x00000008) != 0)) {
+ size += org.apache.pekko.protobufv3.internal.CodedOutputStream
+ .computeInt64Size(4, startedTimestampMillis_);
+ }
+ size += getUnknownFields().getSerializedSize();
+ memoizedSize = size;
+ return size;
+ }
+
+ @java.lang.Override
+ public boolean equals(final java.lang.Object obj) {
+ if (obj == this) {
+ return true;
+ }
+ if (!(obj instanceof org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState)) {
+ return super.equals(obj);
+ }
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState other = (org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState) obj;
+
+ if (hasRevision() != other.hasRevision()) return false;
+ if (hasRevision()) {
+ if (getRevision()
+ != other.getRevision()) return false;
+ }
+ if (hasNumberOfProcesses() != other.hasNumberOfProcesses()) return false;
+ if (hasNumberOfProcesses()) {
+ if (getNumberOfProcesses()
+ != other.getNumberOfProcesses()) return false;
+ }
+ if (hasCompleted() != other.hasCompleted()) return false;
+ if (hasCompleted()) {
+ if (getCompleted()
+ != other.getCompleted()) return false;
+ }
+ if (hasStartedTimestampMillis() != other.hasStartedTimestampMillis()) return false;
+ if (hasStartedTimestampMillis()) {
+ if (getStartedTimestampMillis()
+ != other.getStartedTimestampMillis()) return false;
+ }
+ if (!getUnknownFields().equals(other.getUnknownFields())) return false;
+ return true;
+ }
+
+ @java.lang.Override
+ public int hashCode() {
+ if (memoizedHashCode != 0) {
+ return memoizedHashCode;
+ }
+ int hash = 41;
+ hash = (19 * hash) + getDescriptor().hashCode();
+ if (hasRevision()) {
+ hash = (37 * hash) + REVISION_FIELD_NUMBER;
+ hash = (53 * hash) + org.apache.pekko.protobufv3.internal.Internal.hashLong(
+ getRevision());
+ }
+ if (hasNumberOfProcesses()) {
+ hash = (37 * hash) + NUMBEROFPROCESSES_FIELD_NUMBER;
+ hash = (53 * hash) + getNumberOfProcesses();
+ }
+ if (hasCompleted()) {
+ hash = (37 * hash) + COMPLETED_FIELD_NUMBER;
+ hash = (53 * hash) + org.apache.pekko.protobufv3.internal.Internal.hashBoolean(
+ getCompleted());
+ }
+ if (hasStartedTimestampMillis()) {
+ hash = (37 * hash) + STARTEDTIMESTAMPMILLIS_FIELD_NUMBER;
+ hash = (53 * hash) + org.apache.pekko.protobufv3.internal.Internal.hashLong(
+ getStartedTimestampMillis());
+ }
+ hash = (29 * hash) + getUnknownFields().hashCode();
+ memoizedHashCode = hash;
+ return hash;
+ }
+
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState parseFrom(
+ java.nio.ByteBuffer data)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState parseFrom(
+ java.nio.ByteBuffer data,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data, extensionRegistry);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState parseFrom(
+ org.apache.pekko.protobufv3.internal.ByteString data)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState parseFrom(
+ org.apache.pekko.protobufv3.internal.ByteString data,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data, extensionRegistry);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState parseFrom(byte[] data)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState parseFrom(
+ byte[] data,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data, extensionRegistry);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState parseFrom(java.io.InputStream input)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseWithIOException(PARSER, input);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState parseFrom(
+ java.io.InputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseWithIOException(PARSER, input, extensionRegistry);
+ }
+
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState parseDelimitedFrom(java.io.InputStream input)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseDelimitedWithIOException(PARSER, input);
+ }
+
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState parseDelimitedFrom(
+ java.io.InputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseDelimitedWithIOException(PARSER, input, extensionRegistry);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState parseFrom(
+ org.apache.pekko.protobufv3.internal.CodedInputStream input)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseWithIOException(PARSER, input);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState parseFrom(
+ org.apache.pekko.protobufv3.internal.CodedInputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseWithIOException(PARSER, input, extensionRegistry);
+ }
+
+ @java.lang.Override
+ public Builder newBuilderForType() { return newBuilder(); }
+ public static Builder newBuilder() {
+ return DEFAULT_INSTANCE.toBuilder();
+ }
+ public static Builder newBuilder(org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState prototype) {
+ return DEFAULT_INSTANCE.toBuilder().mergeFrom(prototype);
+ }
+ @java.lang.Override
+ public Builder toBuilder() {
+ return this == DEFAULT_INSTANCE
+ ? new Builder() : new Builder().mergeFrom(this);
+ }
+
+ @java.lang.Override
+ protected Builder newBuilderForType(
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.BuilderParent parent) {
+ Builder builder = new Builder(parent);
+ return builder;
+ }
+ /**
+ * Protobuf type {@code org.apache.pekko.cluster.sharding.typed.DaemonProcessScaleState}
+ */
+ public static final class Builder extends
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.Builder<Builder> implements
+ // @@protoc_insertion_point(builder_implements:org.apache.pekko.cluster.sharding.typed.DaemonProcessScaleState)
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleStateOrBuilder {
+ public static final org.apache.pekko.protobufv3.internal.Descriptors.Descriptor
+ getDescriptor() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_DaemonProcessScaleState_descriptor;
+ }
+
+ @java.lang.Override
+ protected org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable
+ internalGetFieldAccessorTable() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_DaemonProcessScaleState_fieldAccessorTable
+ .ensureFieldAccessorsInitialized(
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState.class, org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState.Builder.class);
+ }
+
+ // Construct using org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState.newBuilder()
+ private Builder() {
+
+ }
+
+ private Builder(
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.BuilderParent parent) {
+ super(parent);
+
+ }
+ @java.lang.Override
+ public Builder clear() {
+ super.clear();
+ bitField0_ = 0;
+ revision_ = 0L;
+ numberOfProcesses_ = 0;
+ completed_ = false;
+ startedTimestampMillis_ = 0L;
+ return this;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.protobufv3.internal.Descriptors.Descriptor
+ getDescriptorForType() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_DaemonProcessScaleState_descriptor;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState getDefaultInstanceForType() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState.getDefaultInstance();
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState build() {
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState result = buildPartial();
+ if (!result.isInitialized()) {
+ throw newUninitializedMessageException(result);
+ }
+ return result;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState buildPartial() {
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState result = new org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState(this);
+ if (bitField0_ != 0) { buildPartial0(result); }
+ onBuilt();
+ return result;
+ }
+
+ private void buildPartial0(org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState result) {
+ int from_bitField0_ = bitField0_;
+ int to_bitField0_ = 0;
+ if (((from_bitField0_ & 0x00000001) != 0)) {
+ result.revision_ = revision_;
+ to_bitField0_ |= 0x00000001;
+ }
+ if (((from_bitField0_ & 0x00000002) != 0)) {
+ result.numberOfProcesses_ = numberOfProcesses_;
+ to_bitField0_ |= 0x00000002;
+ }
+ if (((from_bitField0_ & 0x00000004) != 0)) {
+ result.completed_ = completed_;
+ to_bitField0_ |= 0x00000004;
+ }
+ if (((from_bitField0_ & 0x00000008) != 0)) {
+ result.startedTimestampMillis_ = startedTimestampMillis_;
+ to_bitField0_ |= 0x00000008;
+ }
+ result.bitField0_ |= to_bitField0_;
+ }
+
+ @java.lang.Override
+ public Builder mergeFrom(org.apache.pekko.protobufv3.internal.Message other) {
+ if (other instanceof org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState) {
+ return mergeFrom((org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState)other);
+ } else {
+ super.mergeFrom(other);
+ return this;
+ }
+ }
+
+ public Builder mergeFrom(org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState other) {
+ if (other == org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState.getDefaultInstance()) return this;
+ if (other.hasRevision()) {
+ setRevision(other.getRevision());
+ }
+ if (other.hasNumberOfProcesses()) {
+ setNumberOfProcesses(other.getNumberOfProcesses());
+ }
+ if (other.hasCompleted()) {
+ setCompleted(other.getCompleted());
+ }
+ if (other.hasStartedTimestampMillis()) {
+ setStartedTimestampMillis(other.getStartedTimestampMillis());
+ }
+ this.mergeUnknownFields(other.getUnknownFields());
+ onChanged();
+ return this;
+ }
+
+ @java.lang.Override
+ public final boolean isInitialized() {
+ if (!hasRevision()) {
+ return false;
+ }
+ if (!hasNumberOfProcesses()) {
+ return false;
+ }
+ if (!hasCompleted()) {
+ return false;
+ }
+ if (!hasStartedTimestampMillis()) {
+ return false;
+ }
+ return true;
+ }
+
+ @java.lang.Override
+ public Builder mergeFrom(
+ org.apache.pekko.protobufv3.internal.CodedInputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ if (extensionRegistry == null) {
+ throw new java.lang.NullPointerException();
+ }
+ try {
+ boolean done = false;
+ while (!done) {
+ int tag = input.readTag();
+ switch (tag) {
+ case 0:
+ done = true;
+ break;
+ case 8: {
+ revision_ = input.readInt64();
+ bitField0_ |= 0x00000001;
+ break;
+ } // case 8
+ case 16: {
+ numberOfProcesses_ = input.readInt32();
+ bitField0_ |= 0x00000002;
+ break;
+ } // case 16
+ case 24: {
+ completed_ = input.readBool();
+ bitField0_ |= 0x00000004;
+ break;
+ } // case 24
+ case 32: {
+ startedTimestampMillis_ = input.readInt64();
+ bitField0_ |= 0x00000008;
+ break;
+ } // case 32
+ default: {
+ if (!super.parseUnknownField(input, extensionRegistry, tag)) {
+ done = true; // was an endgroup tag
+ }
+ break;
+ } // default:
+ } // switch (tag)
+ } // while (!done)
+ } catch (org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException e) {
+ throw e.unwrapIOException();
+ } finally {
+ onChanged();
+ } // finally
+ return this;
+ }
+ private int bitField0_;
+
+ private long revision_ ;
+ /**
+ * <code>required int64 revision = 1;</code>
+ * @return Whether the revision field is set.
+ */
+ @java.lang.Override
+ public boolean hasRevision() {
+ return ((bitField0_ & 0x00000001) != 0);
+ }
+ /**
+ * <code>required int64 revision = 1;</code>
+ * @return The revision.
+ */
+ @java.lang.Override
+ public long getRevision() {
+ return revision_;
+ }
+ /**
+ * <code>required int64 revision = 1;</code>
+ * @param value The revision to set.
+ * @return This builder for chaining.
+ */
+ public Builder setRevision(long value) {
+
+ revision_ = value;
+ bitField0_ |= 0x00000001;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>required int64 revision = 1;</code>
+ * @return This builder for chaining.
+ */
+ public Builder clearRevision() {
+ bitField0_ = (bitField0_ & ~0x00000001);
+ revision_ = 0L;
+ onChanged();
+ return this;
+ }
+
+ private int numberOfProcesses_ ;
+ /**
+ * <code>required int32 numberOfProcesses = 2;</code>
+ * @return Whether the numberOfProcesses field is set.
+ */
+ @java.lang.Override
+ public boolean hasNumberOfProcesses() {
+ return ((bitField0_ & 0x00000002) != 0);
+ }
+ /**
+ * <code>required int32 numberOfProcesses = 2;</code>
+ * @return The numberOfProcesses.
+ */
+ @java.lang.Override
+ public int getNumberOfProcesses() {
+ return numberOfProcesses_;
+ }
+ /**
+ * <code>required int32 numberOfProcesses = 2;</code>
+ * @param value The numberOfProcesses to set.
+ * @return This builder for chaining.
+ */
+ public Builder setNumberOfProcesses(int value) {
+
+ numberOfProcesses_ = value;
+ bitField0_ |= 0x00000002;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>required int32 numberOfProcesses = 2;</code>
+ * @return This builder for chaining.
+ */
+ public Builder clearNumberOfProcesses() {
+ bitField0_ = (bitField0_ & ~0x00000002);
+ numberOfProcesses_ = 0;
+ onChanged();
+ return this;
+ }
+
+ private boolean completed_ ;
+ /**
+ * <code>required bool completed = 3;</code>
+ * @return Whether the completed field is set.
+ */
+ @java.lang.Override
+ public boolean hasCompleted() {
+ return ((bitField0_ & 0x00000004) != 0);
+ }
+ /**
+ * <code>required bool completed = 3;</code>
+ * @return The completed.
+ */
+ @java.lang.Override
+ public boolean getCompleted() {
+ return completed_;
+ }
+ /**
+ * <code>required bool completed = 3;</code>
+ * @param value The completed to set.
+ * @return This builder for chaining.
+ */
+ public Builder setCompleted(boolean value) {
+
+ completed_ = value;
+ bitField0_ |= 0x00000004;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>required bool completed = 3;</code>
+ * @return This builder for chaining.
+ */
+ public Builder clearCompleted() {
+ bitField0_ = (bitField0_ & ~0x00000004);
+ completed_ = false;
+ onChanged();
+ return this;
+ }
+
+ private long startedTimestampMillis_ ;
+ /**
+ * <code>required int64 startedTimestampMillis = 4;</code>
+ * @return Whether the startedTimestampMillis field is set.
+ */
+ @java.lang.Override
+ public boolean hasStartedTimestampMillis() {
+ return ((bitField0_ & 0x00000008) != 0);
+ }
+ /**
+ * <code>required int64 startedTimestampMillis = 4;</code>
+ * @return The startedTimestampMillis.
+ */
+ @java.lang.Override
+ public long getStartedTimestampMillis() {
+ return startedTimestampMillis_;
+ }
+ /**
+ * <code>required int64 startedTimestampMillis = 4;</code>
+ * @param value The startedTimestampMillis to set.
+ * @return This builder for chaining.
+ */
+ public Builder setStartedTimestampMillis(long value) {
+
+ startedTimestampMillis_ = value;
+ bitField0_ |= 0x00000008;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>required int64 startedTimestampMillis = 4;</code>
+ * @return This builder for chaining.
+ */
+ public Builder clearStartedTimestampMillis() {
+ bitField0_ = (bitField0_ & ~0x00000008);
+ startedTimestampMillis_ = 0L;
+ onChanged();
+ return this;
+ }
+
+ // @@protoc_insertion_point(builder_scope:org.apache.pekko.cluster.sharding.typed.DaemonProcessScaleState)
+ }
+
+ // @@protoc_insertion_point(class_scope:org.apache.pekko.cluster.sharding.typed.DaemonProcessScaleState)
+ private static final org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState DEFAULT_INSTANCE;
+ static {
+ DEFAULT_INSTANCE = new org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState();
+ }
+
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState getDefaultInstance() {
+ return DEFAULT_INSTANCE;
+ }
+
+ private static final org.apache.pekko.protobufv3.internal.Parser<DaemonProcessScaleState>
+ PARSER = new org.apache.pekko.protobufv3.internal.AbstractParser<DaemonProcessScaleState>() {
+ @java.lang.Override
+ public DaemonProcessScaleState parsePartialFrom(
+ org.apache.pekko.protobufv3.internal.CodedInputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ Builder builder = newBuilder();
+ try {
+ builder.mergeFrom(input, extensionRegistry);
+ } catch (org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException e) {
+ throw e.setUnfinishedMessage(builder.buildPartial());
+ } catch (org.apache.pekko.protobufv3.internal.UninitializedMessageException e) {
+ throw e.asInvalidProtocolBufferException().setUnfinishedMessage(builder.buildPartial());
+ } catch (java.io.IOException e) {
+ throw new org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException(e)
+ .setUnfinishedMessage(builder.buildPartial());
+ }
+ return builder.buildPartial();
+ }
+ };
+
+ public static org.apache.pekko.protobufv3.internal.Parser<DaemonProcessScaleState> parser() {
+ return PARSER;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.protobufv3.internal.Parser<DaemonProcessScaleState> getParserForType() {
+ return PARSER;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.DaemonProcessScaleState getDefaultInstanceForType() {
+ return DEFAULT_INSTANCE;
+ }
+
+ }
+
+ public interface ChangeNumberOfProcessesOrBuilder extends
+ // @@protoc_insertion_point(interface_extends:org.apache.pekko.cluster.sharding.typed.ChangeNumberOfProcesses)
+ org.apache.pekko.protobufv3.internal.MessageOrBuilder {
+
+ /**
+ * <code>required int32 newNumberOfProcesses = 1;</code>
+ * @return Whether the newNumberOfProcesses field is set.
+ */
+ boolean hasNewNumberOfProcesses();
+ /**
+ * <code>required int32 newNumberOfProcesses = 1;</code>
+ * @return The newNumberOfProcesses.
+ */
+ int getNewNumberOfProcesses();
+
+ /**
+ * <code>required string replyTo = 2;</code>
+ * @return Whether the replyTo field is set.
+ */
+ boolean hasReplyTo();
+ /**
+ * <code>required string replyTo = 2;</code>
+ * @return The replyTo.
+ */
+ java.lang.String getReplyTo();
+ /**
+ * <code>required string replyTo = 2;</code>
+ * @return The bytes for replyTo.
+ */
+ org.apache.pekko.protobufv3.internal.ByteString
+ getReplyToBytes();
+ }
+ /**
+ * Protobuf type {@code org.apache.pekko.cluster.sharding.typed.ChangeNumberOfProcesses}
+ */
+ public static final class ChangeNumberOfProcesses extends
+ org.apache.pekko.protobufv3.internal.GeneratedMessage implements
+ // @@protoc_insertion_point(message_implements:org.apache.pekko.cluster.sharding.typed.ChangeNumberOfProcesses)
+ ChangeNumberOfProcessesOrBuilder {
+ private static final long serialVersionUID = 0L;
+ static {
+ org.apache.pekko.protobufv3.internal.RuntimeVersion.validateProtobufGencodeVersion(
+ org.apache.pekko.protobufv3.internal.RuntimeVersion.RuntimeDomain.PUBLIC,
+ /* major= */ 4,
+ /* minor= */ 33,
+ /* patch= */ 0,
+ /* suffix= */ "",
+ "ChangeNumberOfProcesses");
+ }
+ // Use ChangeNumberOfProcesses.newBuilder() to construct.
+ private ChangeNumberOfProcesses(org.apache.pekko.protobufv3.internal.GeneratedMessage.Builder<?> builder) {
+ super(builder);
+ }
+ private ChangeNumberOfProcesses() {
+ replyTo_ = "";
+ }
+
+ public static final org.apache.pekko.protobufv3.internal.Descriptors.Descriptor
+ getDescriptor() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_ChangeNumberOfProcesses_descriptor;
+ }
+
+ @java.lang.Override
+ protected org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable
+ internalGetFieldAccessorTable() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_ChangeNumberOfProcesses_fieldAccessorTable
+ .ensureFieldAccessorsInitialized(
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses.class, org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses.Builder.class);
+ }
+
+ private int bitField0_;
+ public static final int NEWNUMBEROFPROCESSES_FIELD_NUMBER = 1;
+ private int newNumberOfProcesses_ = 0;
+ /**
+ * <code>required int32 newNumberOfProcesses = 1;</code>
+ * @return Whether the newNumberOfProcesses field is set.
+ */
+ @java.lang.Override
+ public boolean hasNewNumberOfProcesses() {
+ return ((bitField0_ & 0x00000001) != 0);
+ }
+ /**
+ * <code>required int32 newNumberOfProcesses = 1;</code>
+ * @return The newNumberOfProcesses.
+ */
+ @java.lang.Override
+ public int getNewNumberOfProcesses() {
+ return newNumberOfProcesses_;
+ }
+
+ public static final int REPLYTO_FIELD_NUMBER = 2;
+ @SuppressWarnings("serial")
+ private volatile java.lang.Object replyTo_ = "";
+ /**
+ * <code>required string replyTo = 2;</code>
+ * @return Whether the replyTo field is set.
+ */
+ @java.lang.Override
+ public boolean hasReplyTo() {
+ return ((bitField0_ & 0x00000002) != 0);
+ }
+ /**
+ * <code>required string replyTo = 2;</code>
+ * @return The replyTo.
+ */
+ @java.lang.Override
+ public java.lang.String getReplyTo() {
+ java.lang.Object ref = replyTo_;
+ if (ref instanceof java.lang.String) {
+ return (java.lang.String) ref;
+ } else {
+ org.apache.pekko.protobufv3.internal.ByteString bs =
+ (org.apache.pekko.protobufv3.internal.ByteString) ref;
+ java.lang.String s = bs.toStringUtf8();
+ if (bs.isValidUtf8()) {
+ replyTo_ = s;
+ }
+ return s;
+ }
+ }
+ /**
+ * <code>required string replyTo = 2;</code>
+ * @return The bytes for replyTo.
+ */
+ @java.lang.Override
+ public org.apache.pekko.protobufv3.internal.ByteString
+ getReplyToBytes() {
+ java.lang.Object ref = replyTo_;
+ if (ref instanceof java.lang.String) {
+ org.apache.pekko.protobufv3.internal.ByteString b =
+ org.apache.pekko.protobufv3.internal.ByteString.copyFromUtf8(
+ (java.lang.String) ref);
+ replyTo_ = b;
+ return b;
+ } else {
+ return (org.apache.pekko.protobufv3.internal.ByteString) ref;
+ }
+ }
+
+ private byte memoizedIsInitialized = -1;
+ @java.lang.Override
+ public final boolean isInitialized() {
+ byte isInitialized = memoizedIsInitialized;
+ if (isInitialized == 1) return true;
+ if (isInitialized == 0) return false;
+
+ if (!hasNewNumberOfProcesses()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
+ if (!hasReplyTo()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
+ memoizedIsInitialized = 1;
+ return true;
+ }
+
+ @java.lang.Override
+ public void writeTo(org.apache.pekko.protobufv3.internal.CodedOutputStream output)
+ throws java.io.IOException {
+ if (((bitField0_ & 0x00000001) != 0)) {
+ output.writeInt32(1, newNumberOfProcesses_);
+ }
+ if (((bitField0_ & 0x00000002) != 0)) {
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.writeString(output, 2, replyTo_);
+ }
+ getUnknownFields().writeTo(output);
+ }
+
+ @java.lang.Override
+ public int getSerializedSize() {
+ int size = memoizedSize;
+ if (size != -1) return size;
+
+ size = 0;
+ if (((bitField0_ & 0x00000001) != 0)) {
+ size += org.apache.pekko.protobufv3.internal.CodedOutputStream
+ .computeInt32Size(1, newNumberOfProcesses_);
+ }
+ if (((bitField0_ & 0x00000002) != 0)) {
+ size += org.apache.pekko.protobufv3.internal.GeneratedMessage.computeStringSize(2, replyTo_);
+ }
+ size += getUnknownFields().getSerializedSize();
+ memoizedSize = size;
+ return size;
+ }
+
+ @java.lang.Override
+ public boolean equals(final java.lang.Object obj) {
+ if (obj == this) {
+ return true;
+ }
+ if (!(obj instanceof org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses)) {
+ return super.equals(obj);
+ }
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses other = (org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses) obj;
+
+ if (hasNewNumberOfProcesses() != other.hasNewNumberOfProcesses()) return false;
+ if (hasNewNumberOfProcesses()) {
+ if (getNewNumberOfProcesses()
+ != other.getNewNumberOfProcesses()) return false;
+ }
+ if (hasReplyTo() != other.hasReplyTo()) return false;
+ if (hasReplyTo()) {
+ if (!getReplyTo()
+ .equals(other.getReplyTo())) return false;
+ }
+ if (!getUnknownFields().equals(other.getUnknownFields())) return false;
+ return true;
+ }
+
+ @java.lang.Override
+ public int hashCode() {
+ if (memoizedHashCode != 0) {
+ return memoizedHashCode;
+ }
+ int hash = 41;
+ hash = (19 * hash) + getDescriptor().hashCode();
+ if (hasNewNumberOfProcesses()) {
+ hash = (37 * hash) + NEWNUMBEROFPROCESSES_FIELD_NUMBER;
+ hash = (53 * hash) + getNewNumberOfProcesses();
+ }
+ if (hasReplyTo()) {
+ hash = (37 * hash) + REPLYTO_FIELD_NUMBER;
+ hash = (53 * hash) + getReplyTo().hashCode();
+ }
+ hash = (29 * hash) + getUnknownFields().hashCode();
+ memoizedHashCode = hash;
+ return hash;
+ }
+
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses parseFrom(
+ java.nio.ByteBuffer data)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses parseFrom(
+ java.nio.ByteBuffer data,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data, extensionRegistry);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses parseFrom(
+ org.apache.pekko.protobufv3.internal.ByteString data)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses parseFrom(
+ org.apache.pekko.protobufv3.internal.ByteString data,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data, extensionRegistry);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses parseFrom(byte[] data)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses parseFrom(
+ byte[] data,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data, extensionRegistry);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses parseFrom(java.io.InputStream input)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseWithIOException(PARSER, input);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses parseFrom(
+ java.io.InputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseWithIOException(PARSER, input, extensionRegistry);
+ }
+
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses parseDelimitedFrom(java.io.InputStream input)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseDelimitedWithIOException(PARSER, input);
+ }
+
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses parseDelimitedFrom(
+ java.io.InputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseDelimitedWithIOException(PARSER, input, extensionRegistry);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses parseFrom(
+ org.apache.pekko.protobufv3.internal.CodedInputStream input)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseWithIOException(PARSER, input);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses parseFrom(
+ org.apache.pekko.protobufv3.internal.CodedInputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseWithIOException(PARSER, input, extensionRegistry);
+ }
+
+ @java.lang.Override
+ public Builder newBuilderForType() { return newBuilder(); }
+ public static Builder newBuilder() {
+ return DEFAULT_INSTANCE.toBuilder();
+ }
+ public static Builder newBuilder(org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses prototype) {
+ return DEFAULT_INSTANCE.toBuilder().mergeFrom(prototype);
+ }
+ @java.lang.Override
+ public Builder toBuilder() {
+ return this == DEFAULT_INSTANCE
+ ? new Builder() : new Builder().mergeFrom(this);
+ }
+
+ @java.lang.Override
+ protected Builder newBuilderForType(
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.BuilderParent parent) {
+ Builder builder = new Builder(parent);
+ return builder;
+ }
+ /**
+ * Protobuf type {@code org.apache.pekko.cluster.sharding.typed.ChangeNumberOfProcesses}
+ */
+ public static final class Builder extends
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.Builder<Builder> implements
+ // @@protoc_insertion_point(builder_implements:org.apache.pekko.cluster.sharding.typed.ChangeNumberOfProcesses)
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcessesOrBuilder {
+ public static final org.apache.pekko.protobufv3.internal.Descriptors.Descriptor
+ getDescriptor() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_ChangeNumberOfProcesses_descriptor;
+ }
+
+ @java.lang.Override
+ protected org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable
+ internalGetFieldAccessorTable() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_ChangeNumberOfProcesses_fieldAccessorTable
+ .ensureFieldAccessorsInitialized(
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses.class, org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses.Builder.class);
+ }
+
+ // Construct using org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses.newBuilder()
+ private Builder() {
+
+ }
+
+ private Builder(
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.BuilderParent parent) {
+ super(parent);
+
+ }
+ @java.lang.Override
+ public Builder clear() {
+ super.clear();
+ bitField0_ = 0;
+ newNumberOfProcesses_ = 0;
+ replyTo_ = "";
+ return this;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.protobufv3.internal.Descriptors.Descriptor
+ getDescriptorForType() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_ChangeNumberOfProcesses_descriptor;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses getDefaultInstanceForType() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses.getDefaultInstance();
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses build() {
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses result = buildPartial();
+ if (!result.isInitialized()) {
+ throw newUninitializedMessageException(result);
+ }
+ return result;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses buildPartial() {
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses result = new org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses(this);
+ if (bitField0_ != 0) { buildPartial0(result); }
+ onBuilt();
+ return result;
+ }
+
+ private void buildPartial0(org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses result) {
+ int from_bitField0_ = bitField0_;
+ int to_bitField0_ = 0;
+ if (((from_bitField0_ & 0x00000001) != 0)) {
+ result.newNumberOfProcesses_ = newNumberOfProcesses_;
+ to_bitField0_ |= 0x00000001;
+ }
+ if (((from_bitField0_ & 0x00000002) != 0)) {
+ result.replyTo_ = replyTo_;
+ to_bitField0_ |= 0x00000002;
+ }
+ result.bitField0_ |= to_bitField0_;
+ }
+
+ @java.lang.Override
+ public Builder mergeFrom(org.apache.pekko.protobufv3.internal.Message other) {
+ if (other instanceof org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses) {
+ return mergeFrom((org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses)other);
+ } else {
+ super.mergeFrom(other);
+ return this;
+ }
+ }
+
+ public Builder mergeFrom(org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses other) {
+ if (other == org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses.getDefaultInstance()) return this;
+ if (other.hasNewNumberOfProcesses()) {
+ setNewNumberOfProcesses(other.getNewNumberOfProcesses());
+ }
+ if (other.hasReplyTo()) {
+ replyTo_ = other.replyTo_;
+ bitField0_ |= 0x00000002;
+ onChanged();
+ }
+ this.mergeUnknownFields(other.getUnknownFields());
+ onChanged();
+ return this;
+ }
+
+ @java.lang.Override
+ public final boolean isInitialized() {
+ if (!hasNewNumberOfProcesses()) {
+ return false;
+ }
+ if (!hasReplyTo()) {
+ return false;
+ }
+ return true;
+ }
+
+ @java.lang.Override
+ public Builder mergeFrom(
+ org.apache.pekko.protobufv3.internal.CodedInputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ if (extensionRegistry == null) {
+ throw new java.lang.NullPointerException();
+ }
+ try {
+ boolean done = false;
+ while (!done) {
+ int tag = input.readTag();
+ switch (tag) {
+ case 0:
+ done = true;
+ break;
+ case 8: {
+ newNumberOfProcesses_ = input.readInt32();
+ bitField0_ |= 0x00000001;
+ break;
+ } // case 8
+ case 18: {
+ replyTo_ = input.readBytes();
+ bitField0_ |= 0x00000002;
+ break;
+ } // case 18
+ default: {
+ if (!super.parseUnknownField(input, extensionRegistry, tag)) {
+ done = true; // was an endgroup tag
+ }
+ break;
+ } // default:
+ } // switch (tag)
+ } // while (!done)
+ } catch (org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException e) {
+ throw e.unwrapIOException();
+ } finally {
+ onChanged();
+ } // finally
+ return this;
+ }
+ private int bitField0_;
+
+ private int newNumberOfProcesses_ ;
+ /**
+ * <code>required int32 newNumberOfProcesses = 1;</code>
+ * @return Whether the newNumberOfProcesses field is set.
+ */
+ @java.lang.Override
+ public boolean hasNewNumberOfProcesses() {
+ return ((bitField0_ & 0x00000001) != 0);
+ }
+ /**
+ * <code>required int32 newNumberOfProcesses = 1;</code>
+ * @return The newNumberOfProcesses.
+ */
+ @java.lang.Override
+ public int getNewNumberOfProcesses() {
+ return newNumberOfProcesses_;
+ }
+ /**
+ * <code>required int32 newNumberOfProcesses = 1;</code>
+ * @param value The newNumberOfProcesses to set.
+ * @return This builder for chaining.
+ */
+ public Builder setNewNumberOfProcesses(int value) {
+
+ newNumberOfProcesses_ = value;
+ bitField0_ |= 0x00000001;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>required int32 newNumberOfProcesses = 1;</code>
+ * @return This builder for chaining.
+ */
+ public Builder clearNewNumberOfProcesses() {
+ bitField0_ = (bitField0_ & ~0x00000001);
+ newNumberOfProcesses_ = 0;
+ onChanged();
+ return this;
+ }
+
+ private java.lang.Object replyTo_ = "";
+ /**
+ * <code>required string replyTo = 2;</code>
+ * @return Whether the replyTo field is set.
+ */
+ public boolean hasReplyTo() {
+ return ((bitField0_ & 0x00000002) != 0);
+ }
+ /**
+ * <code>required string replyTo = 2;</code>
+ * @return The replyTo.
+ */
+ public java.lang.String getReplyTo() {
+ java.lang.Object ref = replyTo_;
+ if (!(ref instanceof java.lang.String)) {
+ org.apache.pekko.protobufv3.internal.ByteString bs =
+ (org.apache.pekko.protobufv3.internal.ByteString) ref;
+ java.lang.String s = bs.toStringUtf8();
+ if (bs.isValidUtf8()) {
+ replyTo_ = s;
+ }
+ return s;
+ } else {
+ return (java.lang.String) ref;
+ }
+ }
+ /**
+ * <code>required string replyTo = 2;</code>
+ * @return The bytes for replyTo.
+ */
+ public org.apache.pekko.protobufv3.internal.ByteString
+ getReplyToBytes() {
+ java.lang.Object ref = replyTo_;
+ if (ref instanceof String) {
+ org.apache.pekko.protobufv3.internal.ByteString b =
+ org.apache.pekko.protobufv3.internal.ByteString.copyFromUtf8(
+ (java.lang.String) ref);
+ replyTo_ = b;
+ return b;
+ } else {
+ return (org.apache.pekko.protobufv3.internal.ByteString) ref;
+ }
+ }
+ /**
+ * <code>required string replyTo = 2;</code>
+ * @param value The replyTo to set.
+ * @return This builder for chaining.
+ */
+ public Builder setReplyTo(
+ java.lang.String value) {
+ if (value == null) { throw new NullPointerException(); }
+ replyTo_ = value;
+ bitField0_ |= 0x00000002;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>required string replyTo = 2;</code>
+ * @return This builder for chaining.
+ */
+ public Builder clearReplyTo() {
+ replyTo_ = getDefaultInstance().getReplyTo();
+ bitField0_ = (bitField0_ & ~0x00000002);
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>required string replyTo = 2;</code>
+ * @param value The bytes for replyTo to set.
+ * @return This builder for chaining.
+ */
+ public Builder setReplyToBytes(
+ org.apache.pekko.protobufv3.internal.ByteString value) {
+ if (value == null) { throw new NullPointerException(); }
+ replyTo_ = value;
+ bitField0_ |= 0x00000002;
+ onChanged();
+ return this;
+ }
+
+ // @@protoc_insertion_point(builder_scope:org.apache.pekko.cluster.sharding.typed.ChangeNumberOfProcesses)
+ }
+
+ // @@protoc_insertion_point(class_scope:org.apache.pekko.cluster.sharding.typed.ChangeNumberOfProcesses)
+ private static final org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses DEFAULT_INSTANCE;
+ static {
+ DEFAULT_INSTANCE = new org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses();
+ }
+
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses getDefaultInstance() {
+ return DEFAULT_INSTANCE;
+ }
+
+ private static final org.apache.pekko.protobufv3.internal.Parser<ChangeNumberOfProcesses>
+ PARSER = new org.apache.pekko.protobufv3.internal.AbstractParser<ChangeNumberOfProcesses>() {
+ @java.lang.Override
+ public ChangeNumberOfProcesses parsePartialFrom(
+ org.apache.pekko.protobufv3.internal.CodedInputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ Builder builder = newBuilder();
+ try {
+ builder.mergeFrom(input, extensionRegistry);
+ } catch (org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException e) {
+ throw e.setUnfinishedMessage(builder.buildPartial());
+ } catch (org.apache.pekko.protobufv3.internal.UninitializedMessageException e) {
+ throw e.asInvalidProtocolBufferException().setUnfinishedMessage(builder.buildPartial());
+ } catch (java.io.IOException e) {
+ throw new org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException(e)
+ .setUnfinishedMessage(builder.buildPartial());
+ }
+ return builder.buildPartial();
+ }
+ };
+
+ public static org.apache.pekko.protobufv3.internal.Parser<ChangeNumberOfProcesses> parser() {
+ return PARSER;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.protobufv3.internal.Parser<ChangeNumberOfProcesses> getParserForType() {
+ return PARSER;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.ChangeNumberOfProcesses getDefaultInstanceForType() {
+ return DEFAULT_INSTANCE;
+ }
+
+ }
+
+ public interface GetNumberOfProcessesOrBuilder extends
+ // @@protoc_insertion_point(interface_extends:org.apache.pekko.cluster.sharding.typed.GetNumberOfProcesses)
+ org.apache.pekko.protobufv3.internal.MessageOrBuilder {
+
+ /**
+ * <code>required string replyTo = 1;</code>
+ * @return Whether the replyTo field is set.
+ */
+ boolean hasReplyTo();
+ /**
+ * <code>required string replyTo = 1;</code>
+ * @return The replyTo.
+ */
+ java.lang.String getReplyTo();
+ /**
+ * <code>required string replyTo = 1;</code>
+ * @return The bytes for replyTo.
+ */
+ org.apache.pekko.protobufv3.internal.ByteString
+ getReplyToBytes();
+ }
+ /**
+ * Protobuf type {@code org.apache.pekko.cluster.sharding.typed.GetNumberOfProcesses}
+ */
+ public static final class GetNumberOfProcesses extends
+ org.apache.pekko.protobufv3.internal.GeneratedMessage implements
+ // @@protoc_insertion_point(message_implements:org.apache.pekko.cluster.sharding.typed.GetNumberOfProcesses)
+ GetNumberOfProcessesOrBuilder {
+ private static final long serialVersionUID = 0L;
+ static {
+ org.apache.pekko.protobufv3.internal.RuntimeVersion.validateProtobufGencodeVersion(
+ org.apache.pekko.protobufv3.internal.RuntimeVersion.RuntimeDomain.PUBLIC,
+ /* major= */ 4,
+ /* minor= */ 33,
+ /* patch= */ 0,
+ /* suffix= */ "",
+ "GetNumberOfProcesses");
+ }
+ // Use GetNumberOfProcesses.newBuilder() to construct.
+ private GetNumberOfProcesses(org.apache.pekko.protobufv3.internal.GeneratedMessage.Builder<?> builder) {
+ super(builder);
+ }
+ private GetNumberOfProcesses() {
+ replyTo_ = "";
+ }
+
+ public static final org.apache.pekko.protobufv3.internal.Descriptors.Descriptor
+ getDescriptor() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcesses_descriptor;
+ }
+
+ @java.lang.Override
+ protected org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable
+ internalGetFieldAccessorTable() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcesses_fieldAccessorTable
+ .ensureFieldAccessorsInitialized(
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses.class, org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses.Builder.class);
+ }
+
+ private int bitField0_;
+ public static final int REPLYTO_FIELD_NUMBER = 1;
+ @SuppressWarnings("serial")
+ private volatile java.lang.Object replyTo_ = "";
+ /**
+ * <code>required string replyTo = 1;</code>
+ * @return Whether the replyTo field is set.
+ */
+ @java.lang.Override
+ public boolean hasReplyTo() {
+ return ((bitField0_ & 0x00000001) != 0);
+ }
+ /**
+ * <code>required string replyTo = 1;</code>
+ * @return The replyTo.
+ */
+ @java.lang.Override
+ public java.lang.String getReplyTo() {
+ java.lang.Object ref = replyTo_;
+ if (ref instanceof java.lang.String) {
+ return (java.lang.String) ref;
+ } else {
+ org.apache.pekko.protobufv3.internal.ByteString bs =
+ (org.apache.pekko.protobufv3.internal.ByteString) ref;
+ java.lang.String s = bs.toStringUtf8();
+ if (bs.isValidUtf8()) {
+ replyTo_ = s;
+ }
+ return s;
+ }
+ }
+ /**
+ * <code>required string replyTo = 1;</code>
+ * @return The bytes for replyTo.
+ */
+ @java.lang.Override
+ public org.apache.pekko.protobufv3.internal.ByteString
+ getReplyToBytes() {
+ java.lang.Object ref = replyTo_;
+ if (ref instanceof java.lang.String) {
+ org.apache.pekko.protobufv3.internal.ByteString b =
+ org.apache.pekko.protobufv3.internal.ByteString.copyFromUtf8(
+ (java.lang.String) ref);
+ replyTo_ = b;
+ return b;
+ } else {
+ return (org.apache.pekko.protobufv3.internal.ByteString) ref;
+ }
+ }
+
+ private byte memoizedIsInitialized = -1;
+ @java.lang.Override
+ public final boolean isInitialized() {
+ byte isInitialized = memoizedIsInitialized;
+ if (isInitialized == 1) return true;
+ if (isInitialized == 0) return false;
+
+ if (!hasReplyTo()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
+ memoizedIsInitialized = 1;
+ return true;
+ }
+
+ @java.lang.Override
+ public void writeTo(org.apache.pekko.protobufv3.internal.CodedOutputStream output)
+ throws java.io.IOException {
+ if (((bitField0_ & 0x00000001) != 0)) {
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.writeString(output, 1, replyTo_);
+ }
+ getUnknownFields().writeTo(output);
+ }
+
+ @java.lang.Override
+ public int getSerializedSize() {
+ int size = memoizedSize;
+ if (size != -1) return size;
+
+ size = 0;
+ if (((bitField0_ & 0x00000001) != 0)) {
+ size += org.apache.pekko.protobufv3.internal.GeneratedMessage.computeStringSize(1, replyTo_);
+ }
+ size += getUnknownFields().getSerializedSize();
+ memoizedSize = size;
+ return size;
+ }
+
+ @java.lang.Override
+ public boolean equals(final java.lang.Object obj) {
+ if (obj == this) {
+ return true;
+ }
+ if (!(obj instanceof org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses)) {
+ return super.equals(obj);
+ }
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses other = (org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses) obj;
+
+ if (hasReplyTo() != other.hasReplyTo()) return false;
+ if (hasReplyTo()) {
+ if (!getReplyTo()
+ .equals(other.getReplyTo())) return false;
+ }
+ if (!getUnknownFields().equals(other.getUnknownFields())) return false;
+ return true;
+ }
+
+ @java.lang.Override
+ public int hashCode() {
+ if (memoizedHashCode != 0) {
+ return memoizedHashCode;
+ }
+ int hash = 41;
+ hash = (19 * hash) + getDescriptor().hashCode();
+ if (hasReplyTo()) {
+ hash = (37 * hash) + REPLYTO_FIELD_NUMBER;
+ hash = (53 * hash) + getReplyTo().hashCode();
+ }
+ hash = (29 * hash) + getUnknownFields().hashCode();
+ memoizedHashCode = hash;
+ return hash;
+ }
+
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses parseFrom(
+ java.nio.ByteBuffer data)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses parseFrom(
+ java.nio.ByteBuffer data,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data, extensionRegistry);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses parseFrom(
+ org.apache.pekko.protobufv3.internal.ByteString data)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses parseFrom(
+ org.apache.pekko.protobufv3.internal.ByteString data,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data, extensionRegistry);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses parseFrom(byte[] data)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses parseFrom(
+ byte[] data,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data, extensionRegistry);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses parseFrom(java.io.InputStream input)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseWithIOException(PARSER, input);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses parseFrom(
+ java.io.InputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseWithIOException(PARSER, input, extensionRegistry);
+ }
+
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses parseDelimitedFrom(java.io.InputStream input)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseDelimitedWithIOException(PARSER, input);
+ }
+
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses parseDelimitedFrom(
+ java.io.InputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseDelimitedWithIOException(PARSER, input, extensionRegistry);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses parseFrom(
+ org.apache.pekko.protobufv3.internal.CodedInputStream input)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseWithIOException(PARSER, input);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses parseFrom(
+ org.apache.pekko.protobufv3.internal.CodedInputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseWithIOException(PARSER, input, extensionRegistry);
+ }
+
+ @java.lang.Override
+ public Builder newBuilderForType() { return newBuilder(); }
+ public static Builder newBuilder() {
+ return DEFAULT_INSTANCE.toBuilder();
+ }
+ public static Builder newBuilder(org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses prototype) {
+ return DEFAULT_INSTANCE.toBuilder().mergeFrom(prototype);
+ }
+ @java.lang.Override
+ public Builder toBuilder() {
+ return this == DEFAULT_INSTANCE
+ ? new Builder() : new Builder().mergeFrom(this);
+ }
+
+ @java.lang.Override
+ protected Builder newBuilderForType(
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.BuilderParent parent) {
+ Builder builder = new Builder(parent);
+ return builder;
+ }
+ /**
+ * Protobuf type {@code org.apache.pekko.cluster.sharding.typed.GetNumberOfProcesses}
+ */
+ public static final class Builder extends
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.Builder<Builder> implements
+ // @@protoc_insertion_point(builder_implements:org.apache.pekko.cluster.sharding.typed.GetNumberOfProcesses)
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesOrBuilder {
+ public static final org.apache.pekko.protobufv3.internal.Descriptors.Descriptor
+ getDescriptor() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcesses_descriptor;
+ }
+
+ @java.lang.Override
+ protected org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable
+ internalGetFieldAccessorTable() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcesses_fieldAccessorTable
+ .ensureFieldAccessorsInitialized(
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses.class, org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses.Builder.class);
+ }
+
+ // Construct using org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses.newBuilder()
+ private Builder() {
+
+ }
+
+ private Builder(
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.BuilderParent parent) {
+ super(parent);
+
+ }
+ @java.lang.Override
+ public Builder clear() {
+ super.clear();
+ bitField0_ = 0;
+ replyTo_ = "";
+ return this;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.protobufv3.internal.Descriptors.Descriptor
+ getDescriptorForType() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcesses_descriptor;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses getDefaultInstanceForType() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses.getDefaultInstance();
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses build() {
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses result = buildPartial();
+ if (!result.isInitialized()) {
+ throw newUninitializedMessageException(result);
+ }
+ return result;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses buildPartial() {
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses result = new org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses(this);
+ if (bitField0_ != 0) { buildPartial0(result); }
+ onBuilt();
+ return result;
+ }
+
+ private void buildPartial0(org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses result) {
+ int from_bitField0_ = bitField0_;
+ int to_bitField0_ = 0;
+ if (((from_bitField0_ & 0x00000001) != 0)) {
+ result.replyTo_ = replyTo_;
+ to_bitField0_ |= 0x00000001;
+ }
+ result.bitField0_ |= to_bitField0_;
+ }
+
+ @java.lang.Override
+ public Builder mergeFrom(org.apache.pekko.protobufv3.internal.Message other) {
+ if (other instanceof org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses) {
+ return mergeFrom((org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses)other);
+ } else {
+ super.mergeFrom(other);
+ return this;
+ }
+ }
+
+ public Builder mergeFrom(org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses other) {
+ if (other == org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses.getDefaultInstance()) return this;
+ if (other.hasReplyTo()) {
+ replyTo_ = other.replyTo_;
+ bitField0_ |= 0x00000001;
+ onChanged();
+ }
+ this.mergeUnknownFields(other.getUnknownFields());
+ onChanged();
+ return this;
+ }
+
+ @java.lang.Override
+ public final boolean isInitialized() {
+ if (!hasReplyTo()) {
+ return false;
+ }
+ return true;
+ }
+
+ @java.lang.Override
+ public Builder mergeFrom(
+ org.apache.pekko.protobufv3.internal.CodedInputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ if (extensionRegistry == null) {
+ throw new java.lang.NullPointerException();
+ }
+ try {
+ boolean done = false;
+ while (!done) {
+ int tag = input.readTag();
+ switch (tag) {
+ case 0:
+ done = true;
+ break;
+ case 10: {
+ replyTo_ = input.readBytes();
+ bitField0_ |= 0x00000001;
+ break;
+ } // case 10
+ default: {
+ if (!super.parseUnknownField(input, extensionRegistry, tag)) {
+ done = true; // was an endgroup tag
+ }
+ break;
+ } // default:
+ } // switch (tag)
+ } // while (!done)
+ } catch (org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException e) {
+ throw e.unwrapIOException();
+ } finally {
+ onChanged();
+ } // finally
+ return this;
+ }
+ private int bitField0_;
+
+ private java.lang.Object replyTo_ = "";
+ /**
+ * <code>required string replyTo = 1;</code>
+ * @return Whether the replyTo field is set.
+ */
+ public boolean hasReplyTo() {
+ return ((bitField0_ & 0x00000001) != 0);
+ }
+ /**
+ * <code>required string replyTo = 1;</code>
+ * @return The replyTo.
+ */
+ public java.lang.String getReplyTo() {
+ java.lang.Object ref = replyTo_;
+ if (!(ref instanceof java.lang.String)) {
+ org.apache.pekko.protobufv3.internal.ByteString bs =
+ (org.apache.pekko.protobufv3.internal.ByteString) ref;
+ java.lang.String s = bs.toStringUtf8();
+ if (bs.isValidUtf8()) {
+ replyTo_ = s;
+ }
+ return s;
+ } else {
+ return (java.lang.String) ref;
+ }
+ }
+ /**
+ * <code>required string replyTo = 1;</code>
+ * @return The bytes for replyTo.
+ */
+ public org.apache.pekko.protobufv3.internal.ByteString
+ getReplyToBytes() {
+ java.lang.Object ref = replyTo_;
+ if (ref instanceof String) {
+ org.apache.pekko.protobufv3.internal.ByteString b =
+ org.apache.pekko.protobufv3.internal.ByteString.copyFromUtf8(
+ (java.lang.String) ref);
+ replyTo_ = b;
+ return b;
+ } else {
+ return (org.apache.pekko.protobufv3.internal.ByteString) ref;
+ }
+ }
+ /**
+ * <code>required string replyTo = 1;</code>
+ * @param value The replyTo to set.
+ * @return This builder for chaining.
+ */
+ public Builder setReplyTo(
+ java.lang.String value) {
+ if (value == null) { throw new NullPointerException(); }
+ replyTo_ = value;
+ bitField0_ |= 0x00000001;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>required string replyTo = 1;</code>
+ * @return This builder for chaining.
+ */
+ public Builder clearReplyTo() {
+ replyTo_ = getDefaultInstance().getReplyTo();
+ bitField0_ = (bitField0_ & ~0x00000001);
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>required string replyTo = 1;</code>
+ * @param value The bytes for replyTo to set.
+ * @return This builder for chaining.
+ */
+ public Builder setReplyToBytes(
+ org.apache.pekko.protobufv3.internal.ByteString value) {
+ if (value == null) { throw new NullPointerException(); }
+ replyTo_ = value;
+ bitField0_ |= 0x00000001;
+ onChanged();
+ return this;
+ }
+
+ // @@protoc_insertion_point(builder_scope:org.apache.pekko.cluster.sharding.typed.GetNumberOfProcesses)
+ }
+
+ // @@protoc_insertion_point(class_scope:org.apache.pekko.cluster.sharding.typed.GetNumberOfProcesses)
+ private static final org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses DEFAULT_INSTANCE;
+ static {
+ DEFAULT_INSTANCE = new org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses();
+ }
+
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses getDefaultInstance() {
+ return DEFAULT_INSTANCE;
+ }
+
+ private static final org.apache.pekko.protobufv3.internal.Parser<GetNumberOfProcesses>
+ PARSER = new org.apache.pekko.protobufv3.internal.AbstractParser<GetNumberOfProcesses>() {
+ @java.lang.Override
+ public GetNumberOfProcesses parsePartialFrom(
+ org.apache.pekko.protobufv3.internal.CodedInputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ Builder builder = newBuilder();
+ try {
+ builder.mergeFrom(input, extensionRegistry);
+ } catch (org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException e) {
+ throw e.setUnfinishedMessage(builder.buildPartial());
+ } catch (org.apache.pekko.protobufv3.internal.UninitializedMessageException e) {
+ throw e.asInvalidProtocolBufferException().setUnfinishedMessage(builder.buildPartial());
+ } catch (java.io.IOException e) {
+ throw new org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException(e)
+ .setUnfinishedMessage(builder.buildPartial());
+ }
+ return builder.buildPartial();
+ }
+ };
+
+ public static org.apache.pekko.protobufv3.internal.Parser<GetNumberOfProcesses> parser() {
+ return PARSER;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.protobufv3.internal.Parser<GetNumberOfProcesses> getParserForType() {
+ return PARSER;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcesses getDefaultInstanceForType() {
+ return DEFAULT_INSTANCE;
+ }
+
+ }
+
+ public interface GetNumberOfProcessesReplyOrBuilder extends
+ // @@protoc_insertion_point(interface_extends:org.apache.pekko.cluster.sharding.typed.GetNumberOfProcessesReply)
+ org.apache.pekko.protobufv3.internal.MessageOrBuilder {
+
+ /**
+ * <code>required int64 revision = 1;</code>
+ * @return Whether the revision field is set.
+ */
+ boolean hasRevision();
+ /**
+ * <code>required int64 revision = 1;</code>
+ * @return The revision.
+ */
+ long getRevision();
+
+ /**
+ * <code>required int32 numberOfProcesses = 2;</code>
+ * @return Whether the numberOfProcesses field is set.
+ */
+ boolean hasNumberOfProcesses();
+ /**
+ * <code>required int32 numberOfProcesses = 2;</code>
+ * @return The numberOfProcesses.
+ */
+ int getNumberOfProcesses();
+
+ /**
+ * <code>required bool rescaleInProgress = 3;</code>
+ * @return Whether the rescaleInProgress field is set.
+ */
+ boolean hasRescaleInProgress();
+ /**
+ * <code>required bool rescaleInProgress = 3;</code>
+ * @return The rescaleInProgress.
+ */
+ boolean getRescaleInProgress();
+
+ /**
+ * <code>required int64 startedTimestampMillis = 4;</code>
+ * @return Whether the startedTimestampMillis field is set.
+ */
+ boolean hasStartedTimestampMillis();
+ /**
+ * <code>required int64 startedTimestampMillis = 4;</code>
+ * @return The startedTimestampMillis.
+ */
+ long getStartedTimestampMillis();
+ }
+ /**
+ * Protobuf type {@code org.apache.pekko.cluster.sharding.typed.GetNumberOfProcessesReply}
+ */
+ public static final class GetNumberOfProcessesReply extends
+ org.apache.pekko.protobufv3.internal.GeneratedMessage implements
+ // @@protoc_insertion_point(message_implements:org.apache.pekko.cluster.sharding.typed.GetNumberOfProcessesReply)
+ GetNumberOfProcessesReplyOrBuilder {
+ private static final long serialVersionUID = 0L;
+ static {
+ org.apache.pekko.protobufv3.internal.RuntimeVersion.validateProtobufGencodeVersion(
+ org.apache.pekko.protobufv3.internal.RuntimeVersion.RuntimeDomain.PUBLIC,
+ /* major= */ 4,
+ /* minor= */ 33,
+ /* patch= */ 0,
+ /* suffix= */ "",
+ "GetNumberOfProcessesReply");
+ }
+ // Use GetNumberOfProcessesReply.newBuilder() to construct.
+ private GetNumberOfProcessesReply(org.apache.pekko.protobufv3.internal.GeneratedMessage.Builder<?> builder) {
+ super(builder);
+ }
+ private GetNumberOfProcessesReply() {
+ }
+
+ public static final org.apache.pekko.protobufv3.internal.Descriptors.Descriptor
+ getDescriptor() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcessesReply_descriptor;
+ }
+
+ @java.lang.Override
+ protected org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable
+ internalGetFieldAccessorTable() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcessesReply_fieldAccessorTable
+ .ensureFieldAccessorsInitialized(
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply.class, org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply.Builder.class);
+ }
+
+ private int bitField0_;
+ public static final int REVISION_FIELD_NUMBER = 1;
+ private long revision_ = 0L;
+ /**
+ * <code>required int64 revision = 1;</code>
+ * @return Whether the revision field is set.
+ */
+ @java.lang.Override
+ public boolean hasRevision() {
+ return ((bitField0_ & 0x00000001) != 0);
+ }
+ /**
+ * <code>required int64 revision = 1;</code>
+ * @return The revision.
+ */
+ @java.lang.Override
+ public long getRevision() {
+ return revision_;
+ }
+
+ public static final int NUMBEROFPROCESSES_FIELD_NUMBER = 2;
+ private int numberOfProcesses_ = 0;
+ /**
+ * <code>required int32 numberOfProcesses = 2;</code>
+ * @return Whether the numberOfProcesses field is set.
+ */
+ @java.lang.Override
+ public boolean hasNumberOfProcesses() {
+ return ((bitField0_ & 0x00000002) != 0);
+ }
+ /**
+ * <code>required int32 numberOfProcesses = 2;</code>
+ * @return The numberOfProcesses.
+ */
+ @java.lang.Override
+ public int getNumberOfProcesses() {
+ return numberOfProcesses_;
+ }
+
+ public static final int RESCALEINPROGRESS_FIELD_NUMBER = 3;
+ private boolean rescaleInProgress_ = false;
+ /**
+ * <code>required bool rescaleInProgress = 3;</code>
+ * @return Whether the rescaleInProgress field is set.
+ */
+ @java.lang.Override
+ public boolean hasRescaleInProgress() {
+ return ((bitField0_ & 0x00000004) != 0);
+ }
+ /**
+ * <code>required bool rescaleInProgress = 3;</code>
+ * @return The rescaleInProgress.
+ */
+ @java.lang.Override
+ public boolean getRescaleInProgress() {
+ return rescaleInProgress_;
+ }
+
+ public static final int STARTEDTIMESTAMPMILLIS_FIELD_NUMBER = 4;
+ private long startedTimestampMillis_ = 0L;
+ /**
+ * <code>required int64 startedTimestampMillis = 4;</code>
+ * @return Whether the startedTimestampMillis field is set.
+ */
+ @java.lang.Override
+ public boolean hasStartedTimestampMillis() {
+ return ((bitField0_ & 0x00000008) != 0);
+ }
+ /**
+ * <code>required int64 startedTimestampMillis = 4;</code>
+ * @return The startedTimestampMillis.
+ */
+ @java.lang.Override
+ public long getStartedTimestampMillis() {
+ return startedTimestampMillis_;
+ }
+
+ private byte memoizedIsInitialized = -1;
+ @java.lang.Override
+ public final boolean isInitialized() {
+ byte isInitialized = memoizedIsInitialized;
+ if (isInitialized == 1) return true;
+ if (isInitialized == 0) return false;
+
+ if (!hasRevision()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
+ if (!hasNumberOfProcesses()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
+ if (!hasRescaleInProgress()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
+ if (!hasStartedTimestampMillis()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
+ memoizedIsInitialized = 1;
+ return true;
+ }
+
+ @java.lang.Override
+ public void writeTo(org.apache.pekko.protobufv3.internal.CodedOutputStream output)
+ throws java.io.IOException {
+ if (((bitField0_ & 0x00000001) != 0)) {
+ output.writeInt64(1, revision_);
+ }
+ if (((bitField0_ & 0x00000002) != 0)) {
+ output.writeInt32(2, numberOfProcesses_);
+ }
+ if (((bitField0_ & 0x00000004) != 0)) {
+ output.writeBool(3, rescaleInProgress_);
+ }
+ if (((bitField0_ & 0x00000008) != 0)) {
+ output.writeInt64(4, startedTimestampMillis_);
+ }
+ getUnknownFields().writeTo(output);
+ }
+
+ @java.lang.Override
+ public int getSerializedSize() {
+ int size = memoizedSize;
+ if (size != -1) return size;
+
+ size = 0;
+ if (((bitField0_ & 0x00000001) != 0)) {
+ size += org.apache.pekko.protobufv3.internal.CodedOutputStream
+ .computeInt64Size(1, revision_);
+ }
+ if (((bitField0_ & 0x00000002) != 0)) {
+ size += org.apache.pekko.protobufv3.internal.CodedOutputStream
+ .computeInt32Size(2, numberOfProcesses_);
+ }
+ if (((bitField0_ & 0x00000004) != 0)) {
+ size += org.apache.pekko.protobufv3.internal.CodedOutputStream
+ .computeBoolSize(3, rescaleInProgress_);
+ }
+ if (((bitField0_ & 0x00000008) != 0)) {
+ size += org.apache.pekko.protobufv3.internal.CodedOutputStream
+ .computeInt64Size(4, startedTimestampMillis_);
+ }
+ size += getUnknownFields().getSerializedSize();
+ memoizedSize = size;
+ return size;
+ }
+
+ @java.lang.Override
+ public boolean equals(final java.lang.Object obj) {
+ if (obj == this) {
+ return true;
+ }
+ if (!(obj instanceof org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply)) {
+ return super.equals(obj);
+ }
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply other = (org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply) obj;
+
+ if (hasRevision() != other.hasRevision()) return false;
+ if (hasRevision()) {
+ if (getRevision()
+ != other.getRevision()) return false;
+ }
+ if (hasNumberOfProcesses() != other.hasNumberOfProcesses()) return false;
+ if (hasNumberOfProcesses()) {
+ if (getNumberOfProcesses()
+ != other.getNumberOfProcesses()) return false;
+ }
+ if (hasRescaleInProgress() != other.hasRescaleInProgress()) return false;
+ if (hasRescaleInProgress()) {
+ if (getRescaleInProgress()
+ != other.getRescaleInProgress()) return false;
+ }
+ if (hasStartedTimestampMillis() != other.hasStartedTimestampMillis()) return false;
+ if (hasStartedTimestampMillis()) {
+ if (getStartedTimestampMillis()
+ != other.getStartedTimestampMillis()) return false;
+ }
+ if (!getUnknownFields().equals(other.getUnknownFields())) return false;
+ return true;
+ }
+
+ @java.lang.Override
+ public int hashCode() {
+ if (memoizedHashCode != 0) {
+ return memoizedHashCode;
+ }
+ int hash = 41;
+ hash = (19 * hash) + getDescriptor().hashCode();
+ if (hasRevision()) {
+ hash = (37 * hash) + REVISION_FIELD_NUMBER;
+ hash = (53 * hash) + org.apache.pekko.protobufv3.internal.Internal.hashLong(
+ getRevision());
+ }
+ if (hasNumberOfProcesses()) {
+ hash = (37 * hash) + NUMBEROFPROCESSES_FIELD_NUMBER;
+ hash = (53 * hash) + getNumberOfProcesses();
+ }
+ if (hasRescaleInProgress()) {
+ hash = (37 * hash) + RESCALEINPROGRESS_FIELD_NUMBER;
+ hash = (53 * hash) + org.apache.pekko.protobufv3.internal.Internal.hashBoolean(
+ getRescaleInProgress());
+ }
+ if (hasStartedTimestampMillis()) {
+ hash = (37 * hash) + STARTEDTIMESTAMPMILLIS_FIELD_NUMBER;
+ hash = (53 * hash) + org.apache.pekko.protobufv3.internal.Internal.hashLong(
+ getStartedTimestampMillis());
+ }
+ hash = (29 * hash) + getUnknownFields().hashCode();
+ memoizedHashCode = hash;
+ return hash;
+ }
+
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply parseFrom(
+ java.nio.ByteBuffer data)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply parseFrom(
+ java.nio.ByteBuffer data,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data, extensionRegistry);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply parseFrom(
+ org.apache.pekko.protobufv3.internal.ByteString data)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply parseFrom(
+ org.apache.pekko.protobufv3.internal.ByteString data,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data, extensionRegistry);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply parseFrom(byte[] data)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply parseFrom(
+ byte[] data,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data, extensionRegistry);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply parseFrom(java.io.InputStream input)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseWithIOException(PARSER, input);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply parseFrom(
+ java.io.InputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseWithIOException(PARSER, input, extensionRegistry);
+ }
+
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply parseDelimitedFrom(java.io.InputStream input)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseDelimitedWithIOException(PARSER, input);
+ }
+
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply parseDelimitedFrom(
+ java.io.InputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseDelimitedWithIOException(PARSER, input, extensionRegistry);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply parseFrom(
+ org.apache.pekko.protobufv3.internal.CodedInputStream input)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseWithIOException(PARSER, input);
+ }
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply parseFrom(
+ org.apache.pekko.protobufv3.internal.CodedInputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return org.apache.pekko.protobufv3.internal.GeneratedMessage
+ .parseWithIOException(PARSER, input, extensionRegistry);
+ }
+
+ @java.lang.Override
+ public Builder newBuilderForType() { return newBuilder(); }
+ public static Builder newBuilder() {
+ return DEFAULT_INSTANCE.toBuilder();
+ }
+ public static Builder newBuilder(org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply prototype) {
+ return DEFAULT_INSTANCE.toBuilder().mergeFrom(prototype);
+ }
+ @java.lang.Override
+ public Builder toBuilder() {
+ return this == DEFAULT_INSTANCE
+ ? new Builder() : new Builder().mergeFrom(this);
+ }
+
+ @java.lang.Override
+ protected Builder newBuilderForType(
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.BuilderParent parent) {
+ Builder builder = new Builder(parent);
+ return builder;
+ }
+ /**
+ * Protobuf type {@code org.apache.pekko.cluster.sharding.typed.GetNumberOfProcessesReply}
+ */
+ public static final class Builder extends
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.Builder<Builder> implements
+ // @@protoc_insertion_point(builder_implements:org.apache.pekko.cluster.sharding.typed.GetNumberOfProcessesReply)
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReplyOrBuilder {
+ public static final org.apache.pekko.protobufv3.internal.Descriptors.Descriptor
+ getDescriptor() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcessesReply_descriptor;
+ }
+
+ @java.lang.Override
+ protected org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable
+ internalGetFieldAccessorTable() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcessesReply_fieldAccessorTable
+ .ensureFieldAccessorsInitialized(
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply.class, org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply.Builder.class);
+ }
+
+ // Construct using org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply.newBuilder()
+ private Builder() {
+
+ }
+
+ private Builder(
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.BuilderParent parent) {
+ super(parent);
+
+ }
+ @java.lang.Override
+ public Builder clear() {
+ super.clear();
+ bitField0_ = 0;
+ revision_ = 0L;
+ numberOfProcesses_ = 0;
+ rescaleInProgress_ = false;
+ startedTimestampMillis_ = 0L;
+ return this;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.protobufv3.internal.Descriptors.Descriptor
+ getDescriptorForType() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcessesReply_descriptor;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply getDefaultInstanceForType() {
+ return org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply.getDefaultInstance();
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply build() {
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply result = buildPartial();
+ if (!result.isInitialized()) {
+ throw newUninitializedMessageException(result);
+ }
+ return result;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply buildPartial() {
+ org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply result = new org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply(this);
+ if (bitField0_ != 0) { buildPartial0(result); }
+ onBuilt();
+ return result;
+ }
+
+ private void buildPartial0(org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply result) {
+ int from_bitField0_ = bitField0_;
+ int to_bitField0_ = 0;
+ if (((from_bitField0_ & 0x00000001) != 0)) {
+ result.revision_ = revision_;
+ to_bitField0_ |= 0x00000001;
+ }
+ if (((from_bitField0_ & 0x00000002) != 0)) {
+ result.numberOfProcesses_ = numberOfProcesses_;
+ to_bitField0_ |= 0x00000002;
+ }
+ if (((from_bitField0_ & 0x00000004) != 0)) {
+ result.rescaleInProgress_ = rescaleInProgress_;
+ to_bitField0_ |= 0x00000004;
+ }
+ if (((from_bitField0_ & 0x00000008) != 0)) {
+ result.startedTimestampMillis_ = startedTimestampMillis_;
+ to_bitField0_ |= 0x00000008;
+ }
+ result.bitField0_ |= to_bitField0_;
+ }
+
+ @java.lang.Override
+ public Builder mergeFrom(org.apache.pekko.protobufv3.internal.Message other) {
+ if (other instanceof org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply) {
+ return mergeFrom((org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply)other);
+ } else {
+ super.mergeFrom(other);
+ return this;
+ }
+ }
+
+ public Builder mergeFrom(org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply other) {
+ if (other == org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply.getDefaultInstance()) return this;
+ if (other.hasRevision()) {
+ setRevision(other.getRevision());
+ }
+ if (other.hasNumberOfProcesses()) {
+ setNumberOfProcesses(other.getNumberOfProcesses());
+ }
+ if (other.hasRescaleInProgress()) {
+ setRescaleInProgress(other.getRescaleInProgress());
+ }
+ if (other.hasStartedTimestampMillis()) {
+ setStartedTimestampMillis(other.getStartedTimestampMillis());
+ }
+ this.mergeUnknownFields(other.getUnknownFields());
+ onChanged();
+ return this;
+ }
+
+ @java.lang.Override
+ public final boolean isInitialized() {
+ if (!hasRevision()) {
+ return false;
+ }
+ if (!hasNumberOfProcesses()) {
+ return false;
+ }
+ if (!hasRescaleInProgress()) {
+ return false;
+ }
+ if (!hasStartedTimestampMillis()) {
+ return false;
+ }
+ return true;
+ }
+
+ @java.lang.Override
+ public Builder mergeFrom(
+ org.apache.pekko.protobufv3.internal.CodedInputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ if (extensionRegistry == null) {
+ throw new java.lang.NullPointerException();
+ }
+ try {
+ boolean done = false;
+ while (!done) {
+ int tag = input.readTag();
+ switch (tag) {
+ case 0:
+ done = true;
+ break;
+ case 8: {
+ revision_ = input.readInt64();
+ bitField0_ |= 0x00000001;
+ break;
+ } // case 8
+ case 16: {
+ numberOfProcesses_ = input.readInt32();
+ bitField0_ |= 0x00000002;
+ break;
+ } // case 16
+ case 24: {
+ rescaleInProgress_ = input.readBool();
+ bitField0_ |= 0x00000004;
+ break;
+ } // case 24
+ case 32: {
+ startedTimestampMillis_ = input.readInt64();
+ bitField0_ |= 0x00000008;
+ break;
+ } // case 32
+ default: {
+ if (!super.parseUnknownField(input, extensionRegistry, tag)) {
+ done = true; // was an endgroup tag
+ }
+ break;
+ } // default:
+ } // switch (tag)
+ } // while (!done)
+ } catch (org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException e) {
+ throw e.unwrapIOException();
+ } finally {
+ onChanged();
+ } // finally
+ return this;
+ }
+ private int bitField0_;
+
+ private long revision_ ;
+ /**
+ * <code>required int64 revision = 1;</code>
+ * @return Whether the revision field is set.
+ */
+ @java.lang.Override
+ public boolean hasRevision() {
+ return ((bitField0_ & 0x00000001) != 0);
+ }
+ /**
+ * <code>required int64 revision = 1;</code>
+ * @return The revision.
+ */
+ @java.lang.Override
+ public long getRevision() {
+ return revision_;
+ }
+ /**
+ * <code>required int64 revision = 1;</code>
+ * @param value The revision to set.
+ * @return This builder for chaining.
+ */
+ public Builder setRevision(long value) {
+
+ revision_ = value;
+ bitField0_ |= 0x00000001;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>required int64 revision = 1;</code>
+ * @return This builder for chaining.
+ */
+ public Builder clearRevision() {
+ bitField0_ = (bitField0_ & ~0x00000001);
+ revision_ = 0L;
+ onChanged();
+ return this;
+ }
+
+ private int numberOfProcesses_ ;
+ /**
+ * <code>required int32 numberOfProcesses = 2;</code>
+ * @return Whether the numberOfProcesses field is set.
+ */
+ @java.lang.Override
+ public boolean hasNumberOfProcesses() {
+ return ((bitField0_ & 0x00000002) != 0);
+ }
+ /**
+ * <code>required int32 numberOfProcesses = 2;</code>
+ * @return The numberOfProcesses.
+ */
+ @java.lang.Override
+ public int getNumberOfProcesses() {
+ return numberOfProcesses_;
+ }
+ /**
+ * <code>required int32 numberOfProcesses = 2;</code>
+ * @param value The numberOfProcesses to set.
+ * @return This builder for chaining.
+ */
+ public Builder setNumberOfProcesses(int value) {
+
+ numberOfProcesses_ = value;
+ bitField0_ |= 0x00000002;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>required int32 numberOfProcesses = 2;</code>
+ * @return This builder for chaining.
+ */
+ public Builder clearNumberOfProcesses() {
+ bitField0_ = (bitField0_ & ~0x00000002);
+ numberOfProcesses_ = 0;
+ onChanged();
+ return this;
+ }
+
+ private boolean rescaleInProgress_ ;
+ /**
+ * <code>required bool rescaleInProgress = 3;</code>
+ * @return Whether the rescaleInProgress field is set.
+ */
+ @java.lang.Override
+ public boolean hasRescaleInProgress() {
+ return ((bitField0_ & 0x00000004) != 0);
+ }
+ /**
+ * <code>required bool rescaleInProgress = 3;</code>
+ * @return The rescaleInProgress.
+ */
+ @java.lang.Override
+ public boolean getRescaleInProgress() {
+ return rescaleInProgress_;
+ }
+ /**
+ * <code>required bool rescaleInProgress = 3;</code>
+ * @param value The rescaleInProgress to set.
+ * @return This builder for chaining.
+ */
+ public Builder setRescaleInProgress(boolean value) {
+
+ rescaleInProgress_ = value;
+ bitField0_ |= 0x00000004;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>required bool rescaleInProgress = 3;</code>
+ * @return This builder for chaining.
+ */
+ public Builder clearRescaleInProgress() {
+ bitField0_ = (bitField0_ & ~0x00000004);
+ rescaleInProgress_ = false;
+ onChanged();
+ return this;
+ }
+
+ private long startedTimestampMillis_ ;
+ /**
+ * <code>required int64 startedTimestampMillis = 4;</code>
+ * @return Whether the startedTimestampMillis field is set.
+ */
+ @java.lang.Override
+ public boolean hasStartedTimestampMillis() {
+ return ((bitField0_ & 0x00000008) != 0);
+ }
+ /**
+ * <code>required int64 startedTimestampMillis = 4;</code>
+ * @return The startedTimestampMillis.
+ */
+ @java.lang.Override
+ public long getStartedTimestampMillis() {
+ return startedTimestampMillis_;
+ }
+ /**
+ * <code>required int64 startedTimestampMillis = 4;</code>
+ * @param value The startedTimestampMillis to set.
+ * @return This builder for chaining.
+ */
+ public Builder setStartedTimestampMillis(long value) {
+
+ startedTimestampMillis_ = value;
+ bitField0_ |= 0x00000008;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>required int64 startedTimestampMillis = 4;</code>
+ * @return This builder for chaining.
+ */
+ public Builder clearStartedTimestampMillis() {
+ bitField0_ = (bitField0_ & ~0x00000008);
+ startedTimestampMillis_ = 0L;
+ onChanged();
+ return this;
+ }
+
+ // @@protoc_insertion_point(builder_scope:org.apache.pekko.cluster.sharding.typed.GetNumberOfProcessesReply)
+ }
+
+ // @@protoc_insertion_point(class_scope:org.apache.pekko.cluster.sharding.typed.GetNumberOfProcessesReply)
+ private static final org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply DEFAULT_INSTANCE;
+ static {
+ DEFAULT_INSTANCE = new org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply();
+ }
+
+ public static org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply getDefaultInstance() {
+ return DEFAULT_INSTANCE;
+ }
+
+ private static final org.apache.pekko.protobufv3.internal.Parser<GetNumberOfProcessesReply>
+ PARSER = new org.apache.pekko.protobufv3.internal.AbstractParser<GetNumberOfProcessesReply>() {
+ @java.lang.Override
+ public GetNumberOfProcessesReply parsePartialFrom(
+ org.apache.pekko.protobufv3.internal.CodedInputStream input,
+ org.apache.pekko.protobufv3.internal.ExtensionRegistryLite extensionRegistry)
+ throws org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException {
+ Builder builder = newBuilder();
+ try {
+ builder.mergeFrom(input, extensionRegistry);
+ } catch (org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException e) {
+ throw e.setUnfinishedMessage(builder.buildPartial());
+ } catch (org.apache.pekko.protobufv3.internal.UninitializedMessageException e) {
+ throw e.asInvalidProtocolBufferException().setUnfinishedMessage(builder.buildPartial());
+ } catch (java.io.IOException e) {
+ throw new org.apache.pekko.protobufv3.internal.InvalidProtocolBufferException(e)
+ .setUnfinishedMessage(builder.buildPartial());
+ }
+ return builder.buildPartial();
+ }
+ };
+
+ public static org.apache.pekko.protobufv3.internal.Parser<GetNumberOfProcessesReply> parser() {
+ return PARSER;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.protobufv3.internal.Parser<GetNumberOfProcessesReply> getParserForType() {
+ return PARSER;
+ }
+
+ @java.lang.Override
+ public org.apache.pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages.GetNumberOfProcessesReply getDefaultInstanceForType() {
+ return DEFAULT_INSTANCE;
+ }
+
+ }
+
private static final org.apache.pekko.protobufv3.internal.Descriptors.Descriptor
internal_static_org_apache_pekko_cluster_sharding_typed_ShardingEnvelope_descriptor;
private static final
org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable
internal_static_org_apache_pekko_cluster_sharding_typed_ShardingEnvelope_fieldAccessorTable;
+ private static final org.apache.pekko.protobufv3.internal.Descriptors.Descriptor
+ internal_static_org_apache_pekko_cluster_sharding_typed_DaemonProcessScaleState_descriptor;
+ private static final
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable
+ internal_static_org_apache_pekko_cluster_sharding_typed_DaemonProcessScaleState_fieldAccessorTable;
+ private static final org.apache.pekko.protobufv3.internal.Descriptors.Descriptor
+ internal_static_org_apache_pekko_cluster_sharding_typed_ChangeNumberOfProcesses_descriptor;
+ private static final
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable
+ internal_static_org_apache_pekko_cluster_sharding_typed_ChangeNumberOfProcesses_fieldAccessorTable;
+ private static final org.apache.pekko.protobufv3.internal.Descriptors.Descriptor
+ internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcesses_descriptor;
+ private static final
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable
+ internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcesses_fieldAccessorTable;
+ private static final org.apache.pekko.protobufv3.internal.Descriptors.Descriptor
+ internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcessesReply_descriptor;
+ private static final
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable
+ internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcessesReply_fieldAccessorTable;
public static org.apache.pekko.protobufv3.internal.Descriptors.FileDescriptor
getDescriptor() {
@@ -824,9 +3632,18 @@ public final class ShardingMessages extends org.apache.pekko.protobufv3.internal
"\n\026ShardingMessages.proto\022\'org.apache.pek" +
"ko.cluster.sharding.typed\032\026ContainerForm" +
"ats.proto\"?\n\020ShardingEnvelope\022\020\n\010entityI" +
- "d\030\001 \002(\t\022\031\n\007message\030\002 \002(\0132\010.PayloadB=\n9or" +
- "g.apache.pekko.cluster.sharding.typed.in" +
- "ternal.protobufH\001"
+ "d\030\001 \002(\t\022\031\n\007message\030\002 \002(\0132\010.Payload\"y\n\027Da" +
+ "emonProcessScaleState\022\020\n\010revision\030\001 \002(\003\022" +
+ "\031\n\021numberOfProcesses\030\002 \002(\005\022\021\n\tcompleted\030" +
+ "\003 \002(\010\022\036\n\026startedTimestampMillis\030\004 \002(\003\"H\n" +
+ "\027ChangeNumberOfProcesses\022\034\n\024newNumberOfP" +
+ "rocesses\030\001 \002(\005\022\017\n\007replyTo\030\002 \002(\t\"\'\n\024GetNu" +
+ "mberOfProcesses\022\017\n\007replyTo\030\001 \002(\t\"\203\001\n\031Get" +
+ "NumberOfProcessesReply\022\020\n\010revision\030\001 \002(\003" +
+ "\022\031\n\021numberOfProcesses\030\002 \002(\005\022\031\n\021rescaleIn" +
+ "Progress\030\003 \002(\010\022\036\n\026startedTimestampMillis" +
+ "\030\004 \002(\003B=\n9org.apache.pekko.cluster.shard" +
+ "ing.typed.internal.protobufH\001"
};
descriptor = org.apache.pekko.protobufv3.internal.Descriptors.FileDescriptor
.internalBuildGeneratedFileFrom(descriptorData,
@@ -839,6 +3656,30 @@ public final class ShardingMessages extends org.apache.pekko.protobufv3.internal
org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable(
internal_static_org_apache_pekko_cluster_sharding_typed_ShardingEnvelope_descriptor,
new java.lang.String[] { "EntityId", "Message", });
+ internal_static_org_apache_pekko_cluster_sharding_typed_DaemonProcessScaleState_descriptor =
+ getDescriptor().getMessageType(1);
+ internal_static_org_apache_pekko_cluster_sharding_typed_DaemonProcessScaleState_fieldAccessorTable = new
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable(
+ internal_static_org_apache_pekko_cluster_sharding_typed_DaemonProcessScaleState_descriptor,
+ new java.lang.String[] { "Revision", "NumberOfProcesses", "Completed", "StartedTimestampMillis", });
+ internal_static_org_apache_pekko_cluster_sharding_typed_ChangeNumberOfProcesses_descriptor =
+ getDescriptor().getMessageType(2);
+ internal_static_org_apache_pekko_cluster_sharding_typed_ChangeNumberOfProcesses_fieldAccessorTable = new
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable(
+ internal_static_org_apache_pekko_cluster_sharding_typed_ChangeNumberOfProcesses_descriptor,
+ new java.lang.String[] { "NewNumberOfProcesses", "ReplyTo", });
+ internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcesses_descriptor =
+ getDescriptor().getMessageType(3);
+ internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcesses_fieldAccessorTable = new
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable(
+ internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcesses_descriptor,
+ new java.lang.String[] { "ReplyTo", });
+ internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcessesReply_descriptor =
+ getDescriptor().getMessageType(4);
+ internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcessesReply_fieldAccessorTable = new
+ org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable(
+ internal_static_org_apache_pekko_cluster_sharding_typed_GetNumberOfProcessesReply_descriptor,
+ new java.lang.String[] { "Revision", "NumberOfProcesses", "RescaleInProgress", "StartedTimestampMillis", });
descriptor.resolveAllFeaturesImmutable();
org.apache.pekko.remote.ContainerFormats.getDescriptor();
}
diff --git a/cluster-sharding-typed/src/main/mima-filters/2.0.x.backwards.excludes/dynamic-scaling.excludes b/cluster-sharding-typed/src/main/mima-filters/2.0.x.backwards.excludes/dynamic-scaling.excludes
new file mode 100644
index 0000000000..c639e5db56
--- /dev/null
+++ b/cluster-sharding-typed/src/main/mima-filters/2.0.x.backwards.excludes/dynamic-scaling.excludes
@@ -0,0 +1,21 @@
+# 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.
+
+# Dynamic scaling of ShardedDaemonProcess instances
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.cluster.sharding.typed.internal.ShardedDaemonProcessImpl$*")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.cluster.sharding.typed.javadsl.ShardedDaemonProcess.initWithContext")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.cluster.sharding.typed.scaladsl.ShardedDaemonProcess.initWithContext")
diff --git a/cluster-sharding-typed/src/main/protobuf/ShardingMessages.proto b/cluster-sharding-typed/src/main/protobuf/ShardingMessages.proto
index 9669413a4d..2653b3251a 100644
--- a/cluster-sharding-typed/src/main/protobuf/ShardingMessages.proto
+++ b/cluster-sharding-typed/src/main/protobuf/ShardingMessages.proto
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2014-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2014-2023 Lightbend Inc. <https://www.lightbend.com>
*/
syntax = "proto2";
@@ -23,3 +23,26 @@ message ShardingEnvelope {
required string entityId = 1;
required Payload message = 2;
}
+
+message DaemonProcessScaleState {
+ required int64 revision = 1;
+ required int32 numberOfProcesses = 2;
+ required bool completed = 3;
+ required int64 startedTimestampMillis = 4;
+}
+
+message ChangeNumberOfProcesses {
+ required int32 newNumberOfProcesses = 1;
+ required string replyTo = 2;
+}
+
+message GetNumberOfProcesses {
+ required string replyTo = 1;
+}
+
+message GetNumberOfProcessesReply {
+ required int64 revision = 1;
+ required int32 numberOfProcesses = 2;
+ required bool rescaleInProgress = 3;
+ required int64 startedTimestampMillis = 4;
+}
diff --git a/cluster-sharding-typed/src/main/resources/reference.conf b/cluster-sharding-typed/src/main/resources/reference.conf
index b7c0a578fa..0bc148a6ee 100644
--- a/cluster-sharding-typed/src/main/resources/reference.conf
+++ b/cluster-sharding-typed/src/main/resources/reference.conf
@@ -48,7 +48,7 @@ pekko.actor {
"org.apache.pekko.cluster.sharding.typed.internal.ShardingSerializer" = 25
}
serialization-bindings {
- "org.apache.pekko.cluster.sharding.typed.ShardingEnvelope" = typed-sharding
+ "org.apache.pekko.cluster.sharding.typed.internal.ClusterShardingTypedSerializable" = typed-sharding
}
}
diff --git a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/ShardedDaemonProcessCommand.scala b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/ShardedDaemonProcessCommand.scala
new file mode 100644
index 0000000000..9fccb9209c
--- /dev/null
+++ b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/ShardedDaemonProcessCommand.scala
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.cluster.sharding.typed
+
+import java.time.Instant
+
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.typed.ActorRef
+import pekko.annotation.DoNotInherit
+import pekko.cluster.sharding.typed.internal.ClusterShardingTypedSerializable
+import pekko.pattern.StatusReply
+
+/**
+ * Commands for interacting with the sharded daemon process
+ *
+ * Not for user extension
+ */
+@DoNotInherit
+trait ShardedDaemonProcessCommand {}
+
+/**
+ * Tell the sharded daemon process to rescale to the given number of processes.
+ *
+ * @param newNumberOfProcesses The number of processes to scale up to
+ * @param replyTo Reply to this actor once scaling is successfully done, or with details if it failed
+ * Note that a successful response may take a long time, depending on how fast
+ * the daemon process actors stop after getting their stop message.
+ */
+final class ChangeNumberOfProcesses(val newNumberOfProcesses: Int, val replyTo: ActorRef[StatusReply[Done]])
+ extends ShardedDaemonProcessCommand
+ with ClusterShardingTypedSerializable {
+
+ override def equals(other: Any): Boolean = other match {
+ case that: ChangeNumberOfProcesses =>
+ newNumberOfProcesses == that.newNumberOfProcesses &&
+ replyTo == that.replyTo
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ val state = Seq(newNumberOfProcesses, replyTo)
+ state.map(_.hashCode()).foldLeft(0)((a, b) => 31 * a + b)
+ }
+}
+
+object ChangeNumberOfProcesses {
+
+ /**
+ * Scala API: Tell the sharded daemon process to rescale to the given number of processes.
+ *
+ * @param newNumberOfProcesses The number of processes to scale up to
+ * @param replyTo Reply to this actor once scaling is successfully done, or with details if it failed
+ * Note that a successful response may take a long time, depending on how fast
+ * the daemon process actors stop after getting their stop message.
+ */
+ def apply(newNumberOfProcesses: Int, replyTo: ActorRef[StatusReply[Done]]): ChangeNumberOfProcesses =
+ new ChangeNumberOfProcesses(newNumberOfProcesses, replyTo)
+
+}
+
+/**
+ * Query the sharded daemon process for the current scale
+ */
+final class GetNumberOfProcesses(val replyTo: ActorRef[NumberOfProcesses])
+ extends ShardedDaemonProcessCommand
+ with ClusterShardingTypedSerializable {
+
+ override def equals(other: Any): Boolean = other match {
+ case that: GetNumberOfProcesses =>
+ replyTo == that.replyTo
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ val state = Seq(replyTo)
+ state.map(_.hashCode()).foldLeft(0)((a, b) => 31 * a + b)
+ }
+}
+
+object GetNumberOfProcesses {
+ def apply(replyTo: ActorRef[NumberOfProcesses]): GetNumberOfProcesses =
+ new GetNumberOfProcesses(replyTo)
+}
+
+/**
+ * Reply for [[GetNumberOfProcesses]]
+ *
+ * Not for user extension
+ */
+@DoNotInherit
+trait NumberOfProcesses {
+
+ def numberOfProcesses: Int
+
+ /**
+ * The timestamp when the change to the current number of processes was initiated. If the number is the initial
+ * number of processes this value is "some time" after cluster startup.
+ */
+ def started: Instant
+
+ def rescaleInProgress: Boolean
+
+ /**
+ * Revision number increased for every re-scale that has been triggered with [[ChangeNumberOfProcesses]]
+ */
+ def revision: Long
+}
diff --git a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/ShardedDaemonProcessSettings.scala b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/ShardedDaemonProcessSettings.scala
index fd940fbaab..7364fd74ea 100644
--- a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/ShardedDaemonProcessSettings.scala
+++ b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/ShardedDaemonProcessSettings.scala
@@ -20,6 +20,8 @@ import scala.jdk.DurationConverters._
import org.apache.pekko
import pekko.actor.typed.ActorSystem
+import pekko.annotation.ApiMayChange
+import pekko.annotation.DoNotInherit
import pekko.annotation.InternalApi
import com.typesafe.config.Config
@@ -119,3 +121,23 @@ final class ShardedDaemonProcessSettings @InternalApi private[pekko] (
keepAliveThrottleInterval)
}
+
+/**
+ * Context with details about the Sharded Daemon Process instance to use when starting it
+ *
+ * Not for user extension
+ */
+@DoNotInherit
+@ApiMayChange
+trait ShardedDaemonProcessContext {
+ def processNumber: Int
+ def totalProcesses: Int
+
+ /**
+ * The revision starts at 0 and each time the number of processes is changed, the revision increases with 1
+ */
+ def revision: Long
+
+ def name: String
+
+}
diff --git a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/ShardingMessageExtractor.scala b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/ShardingMessageExtractor.scala
index f0838fc83f..3f0f4ced92 100644
--- a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/ShardingMessageExtractor.scala
+++ b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/ShardingMessageExtractor.scala
@@ -17,6 +17,7 @@ import scala.annotation.nowarn
import org.apache.pekko
import pekko.actor.{ InvalidMessageException, WrappedMessage }
+import pekko.cluster.sharding.typed.internal.ClusterShardingTypedSerializable
object ShardingMessageExtractor {
@@ -121,6 +122,8 @@ abstract class HashCodeNoEnvelopeMessageExtractor[M](val numberOfShards: Int) ex
* @param message The message to be send to the entity.
* @throws [[pekko.actor.InvalidMessageException]] if message is null.
*/
-final case class ShardingEnvelope[M](entityId: String, message: M) extends WrappedMessage {
+final case class ShardingEnvelope[M](entityId: String, message: M)
+ extends WrappedMessage
+ with ClusterShardingTypedSerializable {
if (message == null) throw InvalidMessageException("[null] is not an allowed message")
}
diff --git a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ClusterShardingTypedSerializable.scala b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ClusterShardingTypedSerializable.scala
new file mode 100644
index 0000000000..69dbf4495b
--- /dev/null
+++ b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ClusterShardingTypedSerializable.scala
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.cluster.sharding.typed.internal
+
+import org.apache.pekko.annotation.InternalApi
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[pekko] trait ClusterShardingTypedSerializable
diff --git a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessCoordinator.scala b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessCoordinator.scala
new file mode 100644
index 0000000000..bda407ffef
--- /dev/null
+++ b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessCoordinator.scala
@@ -0,0 +1,395 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.cluster.sharding.typed.internal
+
+import java.time.Instant
+
+import scala.concurrent.Future
+import scala.concurrent.duration.Duration
+
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.typed.ActorRef
+import pekko.actor.typed.ActorSystem
+import pekko.actor.typed.Behavior
+import pekko.actor.typed.SupervisorStrategy
+import pekko.actor.typed.scaladsl.ActorContext
+import pekko.actor.typed.scaladsl.Behaviors
+import pekko.actor.typed.scaladsl.LoggerOps
+import pekko.actor.typed.scaladsl.TimerScheduler
+import pekko.actor.typed.scaladsl.adapter._
+import pekko.annotation.InternalApi
+import pekko.cluster.ddata.Replicator.ReadAll
+import pekko.cluster.ddata.Replicator.ReadMajorityPlus
+import pekko.cluster.ddata.Replicator.WriteAll
+import pekko.cluster.ddata.Replicator.WriteMajorityPlus
+import pekko.cluster.ddata.typed.scaladsl.DistributedData
+import pekko.cluster.ddata.typed.scaladsl.Replicator
+import pekko.cluster.ddata.typed.scaladsl.ReplicatorMessageAdapter
+import pekko.cluster.sharding.ShardCoordinator
+import pekko.cluster.sharding.typed.ChangeNumberOfProcesses
+import pekko.cluster.sharding.typed.ClusterShardingSettings
+import pekko.cluster.sharding.typed.GetNumberOfProcesses
+import pekko.cluster.sharding.typed.NumberOfProcesses
+import pekko.cluster.sharding.typed.ShardedDaemonProcessCommand
+import pekko.cluster.sharding.typed.ShardedDaemonProcessSettings
+import pekko.cluster.sharding.typed.ShardingEnvelope
+import pekko.cluster.sharding.typed.scaladsl.StartEntity
+import pekko.pattern.StatusReply
+import pekko.stream.KillSwitches
+import pekko.stream.UniqueKillSwitch
+import pekko.stream.scaladsl.Keep
+import pekko.stream.scaladsl.Sink
+import pekko.stream.scaladsl.Source
+
+/**
+ * INTERNAL API
+ *
+ * A ShardedDaemonProcessCoordinator manages dynamic scaling of one setup of ShardedDaemonProcess
+ */
+@InternalApi
+private[pekko] object ShardedDaemonProcessCoordinator {
+
+ private trait InternalMessage extends ShardedDaemonProcessCommand
+
+ private case class InternalGetResponse(rsp: Replicator.GetResponse[ShardedDaemonProcessState])
+ extends InternalMessage
+
+ private case class InternalUpdateResponse(rsp: Replicator.UpdateResponse[ShardedDaemonProcessState])
+ extends InternalMessage
+
+ private case class ShardStopped(shard: String) extends InternalMessage
+
+ private case object Tick extends InternalMessage
+ private case object SendKeepAliveDone extends InternalMessage
+
+ private object ShardStopTimeout extends InternalMessage
+
+ final case class GetNumberOfProcessesReply(
+ numberOfProcesses: Int,
+ started: Instant,
+ rescaleInProgress: Boolean,
+ revision: Long)
+ extends NumberOfProcesses
+ with ClusterShardingTypedSerializable
+
+ def apply[T](
+ settings: ShardedDaemonProcessSettings,
+ shardingSettings: ClusterShardingSettings,
+ initialNumberOfProcesses: Int,
+ daemonProcessName: String,
+ shardingRef: ActorRef[ShardingEnvelope[T]]): Behavior[ShardedDaemonProcessCommand] = {
+ Behaviors
+ .supervise[ShardedDaemonProcessCommand](Behaviors.setup { context =>
+ Behaviors.withTimers {
+ timers =>
+ context.log.debug("ShardedDaemonProcessCoordinator for [{}] starting", daemonProcessName)
+ val key = ShardedDaemonProcessStateKey(daemonProcessName)
+ DistributedData.withReplicatorMessageAdapter[ShardedDaemonProcessCommand, ShardedDaemonProcessState] {
+ replicatorAdapter =>
+ new ShardedDaemonProcessCoordinator(
+ settings,
+ shardingSettings,
+ context,
+ timers,
+ daemonProcessName,
+ shardingRef.toClassic,
+ initialNumberOfProcesses,
+ key,
+ replicatorAdapter).start()
+
+ }
+ }
+ })
+ .onFailure(SupervisorStrategy.restart)
+ }
+}
+
+/**
+ * Rescaling workflow overview:
+ *
+ * 1. stop pinging workers
+ * 2. Write to ddata new size and revision, in progress/not complete
+ * 3. Stop shards for all workers via shard coordinator
+ * 4. Update ddata state, mark rescaling completed
+ * 5. start pinging workers again
+ *
+ * If coordinator moved before completing it will re-trigger workflow from step 2 on start, worst case stopping all
+ * workers before starting them again.
+ */
+private final class ShardedDaemonProcessCoordinator private (
+ settings: ShardedDaemonProcessSettings,
+ shardingSettings: ClusterShardingSettings,
+ context: ActorContext[ShardedDaemonProcessCommand],
+ timers: TimerScheduler[ShardedDaemonProcessCommand],
+ daemonProcessName: String,
+ shardingRef: pekko.actor.ActorRef,
+ initialNumberOfProcesses: Int,
+ key: ShardedDaemonProcessStateKey,
+ replicatorAdapter: ReplicatorMessageAdapter[ShardedDaemonProcessCommand, ShardedDaemonProcessState]) {
+ import ShardedDaemonProcessCoordinator._
+
+ // Stopping shards can take a lot of time, so don't hammer it with retries (even if the
+ // stop shard command is idempotent)
+ private val stopShardsTimeout = shardingSettings.tuningParameters.handOffTimeout / 4
+
+ private val majorityMinCap =
+ context.system.settings.config.getInt("pekko.cluster.sharding.distributed-data.majority-min-cap")
+ private val stateReadConsistency = shardingSettings.tuningParameters.coordinatorStateReadMajorityPlus match {
+ case Int.MaxValue => ReadAll(shardingSettings.tuningParameters.waitingForStateTimeout)
+ case additional =>
+ ReadMajorityPlus(shardingSettings.tuningParameters.waitingForStateTimeout, additional, majorityMinCap)
+ }
+ private val stateWriteConsistency = shardingSettings.tuningParameters.coordinatorStateWriteMajorityPlus match {
+ case Int.MaxValue => WriteAll(shardingSettings.tuningParameters.updatingStateTimeout)
+ case additional =>
+ WriteMajorityPlus(shardingSettings.tuningParameters.updatingStateTimeout, additional, majorityMinCap)
+ }
+
+ private val shardStoppedAdapter = context
+ .messageAdapter[ShardCoordinator.Internal.ShardStopped] {
+ case ShardCoordinator.Internal.ShardStopped(shard) => ShardStopped(shard)
+ }
+ .toClassic
+
+ private def initialState = ShardedDaemonProcessState.initialState(initialNumberOfProcesses)
+
+ def start(): Behavior[ShardedDaemonProcessCommand] = {
+ replicatorAdapter.askGet(
+ replyTo => Replicator.Get(key, stateReadConsistency, replyTo),
+ response => InternalGetResponse(response))
+
+ Behaviors.withStash(100) { stash =>
+ Behaviors.receiveMessage {
+ case InternalGetResponse(success @ Replicator.GetSuccess(`key`)) =>
+ val existingScaleState = success.get(key)
+ if (existingScaleState.completed) {
+ context.log.debugN(
+ "{}: Previous completed state found for Sharded Daemon Process rev [{}] from [{}]",
+ daemonProcessName,
+ existingScaleState.revision,
+ existingScaleState.started)
+ stash.stash(Tick)
+ stash.unstashAll(idle(existingScaleState))
+ } else {
+ context.log.debugN(
+ "{}: Previous non completed state found for Sharded Daemon Process rev [{}] from [{}], retriggering scaling",
+ daemonProcessName,
+ existingScaleState.revision,
+ existingScaleState.started)
+ stopAllShards(existingScaleState, None, initialNumberOfProcesses)
+ }
+
+ case InternalGetResponse(_ @Replicator.NotFound(`key`)) =>
+ context.log.debug("{}: No previous state stored for Sharded Daemon Process", daemonProcessName)
+ stash.stash(Tick)
+ stash.unstashAll(idle(initialState))
+
+ case InternalGetResponse(_ @Replicator.GetFailure(`key`)) =>
+ context.log.info("{}: Failed fetching initial state for Sharded Daemon Process, retrying", daemonProcessName)
+ start()
+
+ case other =>
+ stash.stash(other)
+ Behaviors.same
+ }
+ }
+ }
+
+ private def idle(
+ currentState: ShardedDaemonProcessState,
+ runningThrottledPingKillswitch: Option[UniqueKillSwitch] = None): Behavior[ShardedDaemonProcessCommand] = {
+
+ Behaviors.receiveMessagePartial {
+ case request: ChangeNumberOfProcesses =>
+ if (request.newNumberOfProcesses == currentState.numberOfProcesses) {
+ context.log.info(
+ "{}: Rescale command for Sharded Daemon Process ignored, already running [{}] processes",
+ daemonProcessName,
+ request.newNumberOfProcesses)
+ request.replyTo ! StatusReply.Ack
+ Behaviors.same
+ } else {
+ runningThrottledPingKillswitch.foreach(_.shutdown())
+ val newState = currentState.startScalingTo(request.newNumberOfProcesses)
+ context.log.infoN(
+ "{}: Starting rescaling of Sharded Daemon Process to [{}] processes, rev [{}]",
+ daemonProcessName,
+ request.newNumberOfProcesses,
+ newState.revision)
+ timers.cancel(Tick)
+ prepareRescale(newState, request, currentState.numberOfProcesses)
+ }
+
+ case Tick =>
+ val sortedIdentities =
+ ShardedDaemonProcessId.sortedIdentitiesFor(currentState.revision, currentState.numberOfProcesses)
+ context.log.debugN(
+ "Sending periodic keep alive for Sharded Daemon Process [{}] to [{}] processes (revision [{}]).",
+ daemonProcessName,
+ sortedIdentities.size,
+ currentState.revision)
+ sendKeepAliveMessages(sortedIdentities) match {
+ case None =>
+ timers.startSingleTimer(Tick, settings.keepAliveInterval)
+ Behaviors.same
+ case Some((killSwitch, futureDone)) =>
+ context.pipeToSelf(futureDone)(_ => SendKeepAliveDone)
+ idle(currentState, Some(killSwitch))
+ }
+
+ case SendKeepAliveDone =>
+ timers.startSingleTimer(Tick, settings.keepAliveInterval)
+ Behaviors.same
+
+ case get: GetNumberOfProcesses =>
+ get.replyTo ! GetNumberOfProcessesReply(
+ currentState.numberOfProcesses,
+ currentState.started,
+ !currentState.completed,
+ currentState.revision)
+ Behaviors.same
+ }
+ }
+
+ // we can come here from an explicit rescale request, or because a previous rescale was interrupted by the singleton
+ // moving between nodes and when started anew there was an incomplete scale read from ddata
+ private def prepareRescale(
+ newState: ShardedDaemonProcessState,
+ request: ChangeNumberOfProcesses,
+ previousNumberOfProcesses: Int,
+ updateRetry: Int = 0): Behavior[ShardedDaemonProcessCommand] = {
+ replicatorAdapter.askUpdate(
+ replyTo => Replicator.Update(key, initialState, stateWriteConsistency, replyTo)(_ => newState),
+ response => InternalUpdateResponse(response))
+
+ // wait for update to complete before applying
+ receiveWhileRescaling("prepareRescale", newState) {
+ case InternalUpdateResponse(_ @Replicator.UpdateSuccess(`key`)) =>
+ // this is a bit weird, we use our own state, rather than the result of the update but we also know
+ // there is a single(ton) writer so no surprises here
+ stopAllShards(newState, Some(request), previousNumberOfProcesses)
+
+ case InternalUpdateResponse(_ @Replicator.UpdateTimeout(`key`)) =>
+ // retry
+ if (updateRetry > 5) {
+ context.log.warn2(
+ "{}: Failed updating scale state for sharded daemon process, retrying, retry [{}]",
+ daemonProcessName,
+ updateRetry)
+ } else {
+ context.log.debug("{}: Failed updating scale state for sharded daemon process, retrying", daemonProcessName)
+ }
+ prepareRescale(newState, request, previousNumberOfProcesses, updateRetry + 1)
+
+ }
+ }
+
+ private def stopAllShards(
+ state: ShardedDaemonProcessState,
+ request: Option[ChangeNumberOfProcesses],
+ previousNumberOfProcesses: Int): Behavior[ShardedDaemonProcessCommand] = {
+ val allShards =
+ ShardedDaemonProcessId.allShardsFor(state.revision - 1, previousNumberOfProcesses)
+ shardingRef.tell(ShardCoordinator.Internal.StopShards(allShards), shardStoppedAdapter)
+
+ timers.startSingleTimer(ShardStopTimeout, stopShardsTimeout)
+ waitForShardsStopping(state, request, previousNumberOfProcesses, allShards)
+ }
+
+ private def waitForShardsStopping(
+ state: ShardedDaemonProcessState,
+ request: Option[ChangeNumberOfProcesses],
+ previousNumberOfProcesses: Int,
+ shardsStillRunning: Set[String]): Behavior[ShardedDaemonProcessCommand] = {
+
+ receiveWhileRescaling("waitForShardsStopping", state) {
+ case ShardStopped(shard) =>
+ val newShardsStillRunning = shardsStillRunning - shard
+ if (newShardsStillRunning.isEmpty) {
+ timers.cancel(ShardStopTimeout)
+ rescalingComplete(state, request)
+ } else waitForShardsStopping(state, request, previousNumberOfProcesses, newShardsStillRunning)
+ case ShardStopTimeout =>
+ context.log.debug2("{}: Stopping shards timed out after [{}] retrying", daemonProcessName, stopShardsTimeout)
+ stopAllShards(state, request, previousNumberOfProcesses)
+ }
+ }
+
+ private def rescalingComplete(
+ state: ShardedDaemonProcessState,
+ request: Option[ChangeNumberOfProcesses],
+ ddataWriteRetries: Int = 0): Behavior[ShardedDaemonProcessCommand] = {
+ request.foreach(req => req.replyTo ! StatusReply.Ack)
+ val newState = state.completeScaling()
+ replicatorAdapter.askUpdate(
+ replyTo => Replicator.Update(key, initialState, stateWriteConsistency, replyTo)(_ => newState),
+ response => InternalUpdateResponse(response))
+
+ receiveWhileRescaling("rescalingComplete", state) {
+ case InternalUpdateResponse(_ @Replicator.UpdateSuccess(`key`)) =>
+ context.self ! Tick
+ idle(newState)
+
+ case InternalUpdateResponse(_ @Replicator.UpdateTimeout(`key`)) =>
+ // retry
+ if (ddataWriteRetries < 5) {
+ context.log.debug(
+ "{}: Failed updating scale state for sharded daemon process after completing rescale, retrying",
+ daemonProcessName)
+ } else {
+ context.log.warn2(
+ "{}: Failed updating scale state for sharded daemon process after completing rescale, retry [{}]",
+ daemonProcessName,
+ ddataWriteRetries)
+ }
+ rescalingComplete(newState, None, ddataWriteRetries + 1)
+ }
+ }
+
+ private def receiveWhileRescaling(stepName: String, state: ShardedDaemonProcessState)(
+ pf: PartialFunction[ShardedDaemonProcessCommand, Behavior[ShardedDaemonProcessCommand]])
+ : Behavior[ShardedDaemonProcessCommand] =
+ Behaviors.receiveMessage(pf.orElse {
+ case request: ChangeNumberOfProcesses =>
+ request.replyTo ! StatusReply.error("Rescale in progress, retry later")
+ Behaviors.same
+ case get: GetNumberOfProcesses =>
+ get.replyTo ! GetNumberOfProcessesReply(
+ state.numberOfProcesses,
+ state.started,
+ !state.completed,
+ state.revision)
+ Behaviors.same
+
+ case msg =>
+ context.log.debugN("{}: Unexpected message in step {}: {}", daemonProcessName, stepName, msg.getClass.getName)
+ Behaviors.same
+ })
+
+ private def sendKeepAliveMessages(sortedIdentities: Vector[String]): Option[(UniqueKillSwitch, Future[Done])] = {
+ if (settings.keepAliveThrottleInterval == Duration.Zero) {
+ sortedIdentities.foreach(id => shardingRef ! StartEntity(id))
+ None
+ } else {
+ implicit val system: ActorSystem[_] = context.system
+ Some(
+ Source(sortedIdentities)
+ .viaMat(KillSwitches.single)(Keep.right)
+ .throttle(1, settings.keepAliveThrottleInterval)
+ .toMat(Sink.foreach(id => shardingRef ! StartEntity(id)))(Keep.both)
+ .run())
+ }
+ }
+}
diff --git a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessId.scala b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessId.scala
new file mode 100644
index 0000000000..5310a9f286
--- /dev/null
+++ b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessId.scala
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.cluster.sharding.typed.internal
+
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.cluster.sharding.typed.ShardingEnvelope
+import pekko.cluster.sharding.typed.ShardingMessageExtractor
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[pekko] object ShardedDaemonProcessId {
+
+ /*
+ * The entity id used for sharded daemon process has the format: [revision]|[total-count]|[this-process-id]
+ *
+ * Pingers repeatedly ping each worker id for a given number of processes to keep them alive. When rescaling
+ * to a new number of workers the revision is increased so that we can detect pings for old processes and avoid
+ * starting old workers anew.
+ */
+ val Separator = '|'
+
+ def decodeEntityId(id: String, initialNumberOfProcesses: Int) =
+ id.split(Separator) match {
+ case Array(rev, count, n) => DecodedId(rev.toLong, count.toInt, n.toInt)
+ case Array(n) =>
+ DecodedId(0L, initialNumberOfProcesses, n.toInt) // ping from old/supportsRescale=false node during rolling upgrade
+ case _ => throw new IllegalArgumentException(s"Unexpected id for sharded daemon process: '$id'")
+ }
+
+ final case class DecodedId(revision: Long, totalCount: Int, processNumber: Int) {
+ def encodeEntityId(): String =
+ if (revision == 0L) {
+ // to safely do rolling updates, we encode revision 0 using the old static scheme
+ processNumber.toString
+ } else s"$revision$Separator$totalCount$Separator$processNumber"
+ }
+
+ final class MessageExtractor[T] extends ShardingMessageExtractor[ShardingEnvelope[T], T] {
+ def entityId(message: ShardingEnvelope[T]): String = message match {
+ case ShardingEnvelope(id, _) => id
+ }
+
+ // use process n for shard id
+ def shardId(entityId: String): String =
+ entityId.split(Separator) match {
+ case Array(_, _, id) => id
+ case Array(id) => id // ping from old/supportsRescale=false node during rolling upgrade
+ case id => throw new IllegalArgumentException(s"Unexpected id for sharded daemon process: '$id'")
+ }
+
+ def unwrapMessage(message: ShardingEnvelope[T]): T = message.message
+ }
+
+ def sortedIdentitiesFor(revision: Long, numberOfProcesses: Int): Vector[String] = (0 until numberOfProcesses).map(n =>
+ DecodedId(revision, numberOfProcesses, n).encodeEntityId()).toVector.sorted
+
+ def allShardsFor(revision: Long, numberOfProcesses: Int): Set[String] = {
+ val messageExtractor = new MessageExtractor[Unit]()
+ sortedIdentitiesFor(revision, numberOfProcesses).map(messageExtractor.shardId).toSet
+ }
+
+}
diff --git a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessImpl.scala b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessImpl.scala
index f8b17f56f2..fed10e2ab4 100644
--- a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessImpl.scala
+++ b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessImpl.scala
@@ -17,120 +17,30 @@ import java.util.Optional
import java.util.function.IntFunction
import scala.jdk.OptionConverters._
-import scala.concurrent.Future
-import scala.concurrent.duration.Duration
import scala.reflect.ClassTag
import org.apache.pekko
-import pekko.Done
import pekko.actor.typed.ActorRef
import pekko.actor.typed.ActorSystem
import pekko.actor.typed.Behavior
-import pekko.actor.typed.scaladsl.Behaviors
-import pekko.actor.typed.scaladsl.LoggerOps
import pekko.annotation.InternalApi
-import pekko.cluster.MemberStatus
+import pekko.cluster.ddata.typed.scaladsl.DistributedData
import pekko.cluster.sharding.ShardCoordinator.ShardAllocationStrategy
import pekko.cluster.sharding.ShardRegion.EntityId
import pekko.cluster.sharding.typed.ClusterShardingSettings
-import pekko.cluster.sharding.typed.ClusterShardingSettings.{ RememberEntitiesStoreModeDData, StateStoreModeDData }
+import pekko.cluster.sharding.typed.ClusterShardingSettings.RememberEntitiesStoreModeDData
+import pekko.cluster.sharding.typed.ClusterShardingSettings.StateStoreModeDData
+import pekko.cluster.sharding.typed.ShardedDaemonProcessCommand
+import pekko.cluster.sharding.typed.ShardedDaemonProcessContext
import pekko.cluster.sharding.typed.ShardedDaemonProcessSettings
-import pekko.cluster.sharding.typed.ShardingEnvelope
-import pekko.cluster.sharding.typed.ShardingMessageExtractor
import pekko.cluster.sharding.typed.javadsl
import pekko.cluster.sharding.typed.scaladsl
import pekko.cluster.sharding.typed.scaladsl.ClusterSharding
import pekko.cluster.sharding.typed.scaladsl.Entity
import pekko.cluster.sharding.typed.scaladsl.EntityTypeKey
-import pekko.cluster.sharding.typed.scaladsl.StartEntity
-import pekko.cluster.typed.Cluster
-import pekko.cluster.typed.SelfUp
-import pekko.cluster.typed.Subscribe
-import pekko.stream.scaladsl.Source
-
-/**
- * INTERNAL API
- */
-@InternalApi
-private[pekko] object ShardedDaemonProcessImpl {
-
- object KeepAlivePinger {
- sealed trait Event
- private case object Tick extends Event
- private case object SendKeepAliveDone extends Event
-
- def apply[T](
- settings: ShardedDaemonProcessSettings,
- name: String,
- identities: Set[EntityId],
- shardingRef: ActorRef[ShardingEnvelope[T]]): Behavior[Event] = {
- val sortedIdentities = identities.toVector.sorted
-
- def sendKeepAliveMessages()(implicit sys: ActorSystem[_]): Future[Done] = {
- if (settings.keepAliveThrottleInterval == Duration.Zero) {
- sortedIdentities.foreach(id => shardingRef ! StartEntity(id))
- Future.successful(Done)
- } else {
- Source(sortedIdentities).throttle(1, settings.keepAliveThrottleInterval).runForeach { id =>
- shardingRef ! StartEntity(id)
- }
- }
- }
-
- Behaviors.setup[Event] { context =>
- implicit val system: ActorSystem[_] = context.system
- val cluster = Cluster(system)
-
- if (cluster.selfMember.status == MemberStatus.Up)
- context.self ! Tick
- else
- cluster.subscriptions ! Subscribe(context.messageAdapter[SelfUp](_ => Tick), classOf[SelfUp])
-
- def isActive(): Boolean = {
- val members = settings.role match {
- case None => cluster.state.members
- case Some(role) => cluster.state.members.filter(_.roles.contains(role))
- }
- // members are sorted so this is deterministic (the same) on all nodes
- members.take(settings.keepAliveFromNumberOfNodes).contains(cluster.selfMember)
- }
-
- Behaviors.withTimers { timers =>
- Behaviors.receiveMessage {
- case Tick =>
- if (isActive()) {
- context.log.debug2(
- s"Sending periodic keep alive for Sharded Daemon Process [{}] to [{}] processes.",
- name,
- sortedIdentities.size)
- context.pipeToSelf(sendKeepAliveMessages()) { _ =>
- SendKeepAliveDone
- }
- } else {
- timers.startSingleTimer(Tick, settings.keepAliveInterval)
- }
- Behaviors.same
- case SendKeepAliveDone =>
- timers.startSingleTimer(Tick, settings.keepAliveInterval)
- Behaviors.same
- }
- }
- }
- }
-
- }
-
- final class MessageExtractor[T] extends ShardingMessageExtractor[ShardingEnvelope[T], T] {
- def entityId(message: ShardingEnvelope[T]): String = message match {
- case ShardingEnvelope(id, _) => id
- }
-
- def shardId(entityId: String): String = entityId
-
- def unwrapMessage(message: ShardingEnvelope[T]): T = message.message
- }
-
-}
+import pekko.cluster.typed.ClusterSingleton
+import pekko.cluster.typed.ClusterSingletonSettings
+import pekko.cluster.typed.SingletonActor
/**
* INTERNAL API
@@ -140,7 +50,15 @@ private[pekko] final class ShardedDaemonProcessImpl(system: ActorSystem[_])
extends javadsl.ShardedDaemonProcess
with scaladsl.ShardedDaemonProcess {
- import ShardedDaemonProcessImpl._
+ import ShardedDaemonProcessId._
+ import ShardedDaemonProcessState.verifyRevisionBeforeStarting
+
+ private case class ShardedDaemonProcessContextImpl(
+ processNumber: Int,
+ totalProcesses: Int,
+ name: String,
+ revision: Long)
+ extends ShardedDaemonProcessContext
def init[T](name: String, numberOfInstances: Int, behaviorFactory: Int => Behavior[T])(
implicit classTag: ClassTag[T]): Unit =
@@ -151,7 +69,7 @@ private[pekko] final class ShardedDaemonProcessImpl(system: ActorSystem[_])
init(name, numberOfInstances, behaviorFactory, ShardedDaemonProcessSettings(system), Some(stopMessage), None)(
classTag)
- def init[T](
+ override def init[T](
name: String,
numberOfInstances: Int,
behaviorFactory: Int => Behavior[T],
@@ -159,19 +77,75 @@ private[pekko] final class ShardedDaemonProcessImpl(system: ActorSystem[_])
stopMessage: Option[T])(implicit classTag: ClassTag[T]): Unit =
init(name, numberOfInstances, behaviorFactory, settings, stopMessage, None)
- def init[T](
+ override def init[T](
name: String,
numberOfInstances: Int,
behaviorFactory: Int => Behavior[T],
settings: ShardedDaemonProcessSettings,
stopMessage: Option[T],
- shardAllocationStrategy: Option[ShardAllocationStrategy])(implicit classTag: ClassTag[T]): Unit = {
+ shardAllocationStrategy: Option[ShardAllocationStrategy])(implicit classTag: ClassTag[T]): Unit =
+ internalInitWithContext(
+ name,
+ numberOfInstances,
+ context => behaviorFactory(context.processNumber),
+ Some(settings),
+ stopMessage,
+ shardAllocationStrategy,
+ supportsRescale = false)
+
+ override def initWithContext[T](
+ name: EntityId,
+ initialNumberOfInstances: Int,
+ behaviorFactory: ShardedDaemonProcessContext => Behavior[T])(
+ implicit classTag: ClassTag[T]): ActorRef[ShardedDaemonProcessCommand] =
+ internalInitWithContext(name, initialNumberOfInstances, behaviorFactory, None, None, None, true)
+
+ override def initWithContext[T](
+ name: EntityId,
+ initialNumberOfInstances: Int,
+ behaviorFactory: ShardedDaemonProcessContext => Behavior[T],
+ settings: ShardedDaemonProcessSettings,
+ stopMessage: T)(implicit classTag: ClassTag[T]): ActorRef[ShardedDaemonProcessCommand] =
+ internalInitWithContext(
+ name,
+ initialNumberOfInstances,
+ behaviorFactory,
+ Some(settings),
+ Some(stopMessage),
+ None,
+ true)
+
+ override def initWithContext[T](
+ name: String,
+ numberOfInstances: Int,
+ behaviorFactory: ShardedDaemonProcessContext => Behavior[T],
+ settings: ShardedDaemonProcessSettings,
+ stopMessage: Option[T],
+ shardAllocationStrategy: Option[ShardAllocationStrategy])(
+ implicit classTag: ClassTag[T]): ActorRef[ShardedDaemonProcessCommand] =
+ internalInitWithContext(
+ name,
+ numberOfInstances,
+ behaviorFactory,
+ Some(settings),
+ stopMessage,
+ shardAllocationStrategy,
+ supportsRescale = true)
+
+ private def internalInitWithContext[T](
+ name: String,
+ numberOfInstances: Int,
+ behaviorFactory: ShardedDaemonProcessContext => Behavior[T],
+ maybeSettings: Option[ShardedDaemonProcessSettings],
+ stopMessage: Option[T],
+ shardAllocationStrategy: Option[ShardAllocationStrategy],
+ supportsRescale: Boolean)(implicit classTag: ClassTag[T]): ActorRef[ShardedDaemonProcessCommand] = {
+ val settings = maybeSettings.getOrElse(ShardedDaemonProcessSettings(system))
val entityTypeKey = EntityTypeKey[T](s"sharded-daemon-process-$name")
// One shard per actor identified by the numeric id encoded in the entity id
val numberOfShards = numberOfInstances
- val entityIds = (0 until numberOfInstances).map(_.toString)
val shardingSettings = {
val shardingBaseSettings =
@@ -200,32 +174,43 @@ private[pekko] final class ShardedDaemonProcessImpl(system: ActorSystem[_])
shardingBaseSettings.leaseSettings)
}
- val nodeRoles = Cluster(system).selfMember.roles
- if (shardingSettings.role.forall(nodeRoles)) {
- val entity = Entity(entityTypeKey)(ctx => behaviorFactory(ctx.entityId.toInt))
- .withSettings(shardingSettings)
- .withMessageExtractor(new MessageExtractor)
+ val entity = Entity(entityTypeKey) { ctx =>
+ val decodedId = decodeEntityId(ctx.entityId, initialNumberOfProcesses = numberOfInstances)
+ val sdContext =
+ ShardedDaemonProcessContextImpl(decodedId.processNumber, decodedId.totalCount, name, decodedId.revision)
+ if (supportsRescale) verifyRevisionBeforeStarting(behaviorFactory)(sdContext)
+ else
+ behaviorFactory(sdContext)
+ }.withSettings(shardingSettings).withMessageExtractor(new MessageExtractor())
+
+ val entityWithStop = stopMessage match {
+ case Some(stop) => entity.withStopMessage(stop)
+ case None => entity
+ }
+
+ val entityWithShardAllocationStrategy = shardAllocationStrategy match {
+ case Some(strategy) => entityWithStop.withAllocationStrategy(strategy)
+ case None => entityWithStop
+ }
- val entityWithStop = stopMessage match {
- case Some(stop) => entity.withStopMessage(stop)
- case None => entity
- }
+ val shardingRef = ClusterSharding(system).init(entityWithShardAllocationStrategy)
- val entityWithShardAllocationStrategy = shardAllocationStrategy match {
- case Some(strategy) => entityWithStop.withAllocationStrategy(strategy)
- case None => entityWithStop
- }
+ // started on all nodes even if using roles to be able to share the default replicator
+ DistributedData(system).replicator
- val shardingRef = ClusterSharding(system).init(entityWithShardAllocationStrategy)
+ var singletonSettings =
+ ClusterSingletonSettings(system)
+ settings.role.foreach(role => singletonSettings = singletonSettings.withRole(role))
+ val singleton =
+ SingletonActor(
+ ShardedDaemonProcessCoordinator(settings, shardingSettings, numberOfInstances, name, shardingRef),
+ s"ShardedDaemonProcessCoordinator-$name").withSettings(singletonSettings)
- system.systemActorOf(
- KeepAlivePinger(settings, name, entityIds.toSet, shardingRef),
- s"ShardedDaemonProcessKeepAlive-$name")
- }
+ ClusterSingleton(system).init(singleton)
}
// Java API
- def init[T](
+ override def init[T](
messageClass: Class[T],
name: String,
numberOfInstances: Int,
@@ -246,7 +231,7 @@ private[pekko] final class ShardedDaemonProcessImpl(system: ActorSystem[_])
Some(stopMessage),
None)(ClassTag(messageClass))
- def init[T](
+ override def init[T](
messageClass: Class[T],
name: String,
numberOfInstances: Int,
@@ -255,7 +240,7 @@ private[pekko] final class ShardedDaemonProcessImpl(system: ActorSystem[_])
stopMessage: Optional[T]): Unit =
init(name, numberOfInstances, n => behaviorFactory(n), settings, stopMessage.toScala, None)(ClassTag(messageClass))
- def init[T](
+ override def init[T](
messageClass: Class[T],
name: String,
numberOfInstances: Int,
@@ -270,4 +255,49 @@ private[pekko] final class ShardedDaemonProcessImpl(system: ActorSystem[_])
settings,
stopMessage.toScala,
shardAllocationStrategy.toScala)(ClassTag(messageClass))
+
+ override def initWithContext[T](
+ messageClass: Class[T],
+ name: String,
+ initialNumberOfInstances: Int,
+ behaviorFactory: java.util.function.Function[ShardedDaemonProcessContext, Behavior[T]])
+ : ActorRef[ShardedDaemonProcessCommand] = {
+ val classTag = ClassTag[T](messageClass)
+ internalInitWithContext[T](name, initialNumberOfInstances, behaviorFactory.apply, None, None, None, true)(classTag)
+ }
+
+ override def initWithContext[T](
+ messageClass: Class[T],
+ name: String,
+ initialNumberOfInstances: Int,
+ behaviorFactory: java.util.function.Function[ShardedDaemonProcessContext, Behavior[T]],
+ settings: ShardedDaemonProcessSettings,
+ stopMessage: Optional[T]): ActorRef[ShardedDaemonProcessCommand] =
+ initWithContext[T](
+ messageClass,
+ name,
+ initialNumberOfInstances,
+ behaviorFactory,
+ settings,
+ stopMessage,
+ Optional.empty[ShardAllocationStrategy]())
+
+ override def initWithContext[T](
+ messageClass: Class[T],
+ name: String,
+ initialNumberOfInstances: Int,
+ behaviorFactory: java.util.function.Function[ShardedDaemonProcessContext, Behavior[T]],
+ settings: ShardedDaemonProcessSettings,
+ stopMessage: Optional[T],
+ shardAllocationStrategy: Optional[ShardAllocationStrategy]): ActorRef[ShardedDaemonProcessCommand] = {
+ val classTag = ClassTag[T](messageClass)
+ internalInitWithContext(
+ name,
+ initialNumberOfInstances,
+ behaviorFactory.apply,
+ Some(settings),
+ stopMessage.toScala,
+ shardAllocationStrategy.toScala,
+ supportsRescale = true)(classTag)
+ }
}
diff --git a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessState.scala b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessState.scala
new file mode 100644
index 0000000000..6febc38d35
--- /dev/null
+++ b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessState.scala
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.cluster.sharding.typed.internal
+
+import java.time.Instant
+
+import org.apache.pekko
+import pekko.actor.typed.Behavior
+import pekko.actor.typed.scaladsl.Behaviors
+import pekko.actor.typed.scaladsl.LoggerOps
+import pekko.annotation.InternalApi
+import pekko.cluster.ddata.Key
+import pekko.cluster.ddata.ReplicatedData
+import pekko.cluster.ddata.typed.scaladsl.DistributedData
+import pekko.cluster.ddata.typed.scaladsl.Replicator
+import pekko.cluster.sharding.typed.ShardedDaemonProcessContext
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[pekko] final case class ShardedDaemonProcessState(
+ revision: Long,
+ numberOfProcesses: Int,
+ completed: Boolean,
+ started: Instant)
+ extends ReplicatedData
+ with ClusterShardingTypedSerializable {
+ type T = ShardedDaemonProcessState
+
+ override def merge(that: ShardedDaemonProcessState): ShardedDaemonProcessState =
+ if (this.revision == that.revision) {
+ if (this.completed) this
+ else that
+ } else if (this.revision > that.revision)
+ this
+ else that
+
+ def startScalingTo(newNumberOfProcesses: Int): ShardedDaemonProcessState =
+ copy(revision = revision + 1L, completed = false, numberOfProcesses = newNumberOfProcesses, started = Instant.now())
+ def completeScaling(): ShardedDaemonProcessState = copy(completed = true)
+
+}
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[pekko] final case class ShardedDaemonProcessStateKey(_id: String)
+ extends Key[ShardedDaemonProcessState](_id)
+ with ClusterShardingTypedSerializable {
+ override def withId(newId: Key.KeyId): ShardedDaemonProcessStateKey =
+ ShardedDaemonProcessStateKey(newId)
+}
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[pekko] object ShardedDaemonProcessState {
+
+ val startRevision = 0L
+
+ def initialState(initialNumberOfProcesses: Int) =
+ ShardedDaemonProcessState(
+ revision = ShardedDaemonProcessState.startRevision,
+ numberOfProcesses = initialNumberOfProcesses,
+ completed = true,
+ // not quite correct but also not important, only informational
+ started = Instant.now())
+
+ def verifyRevisionBeforeStarting[T](
+ behaviorFactory: ShardedDaemonProcessContext => Behavior[T]): ShardedDaemonProcessContext => Behavior[T] = {
+ sdpContext =>
+ Behaviors.setup { context =>
+ val revision = sdpContext.revision
+
+ if (revision == -1) {
+ context.log.debug2(
+ "{}: Ping from old non-rescaling node during rolling upgrade, not starting worker [{}]",
+ sdpContext.name,
+ sdpContext.processNumber)
+ Behaviors.stopped
+ } else {
+ val key = ShardedDaemonProcessStateKey(sdpContext.name)
+ context.log.debug2(
+ "{}: Deferred start of worker to verify its revision [{}] is the latest",
+ sdpContext.name,
+ revision)
+
+ // we can't anyway turn reply into T so no need for the usual adapter
+ val distributedData = DistributedData(context.system)
+ distributedData.replicator ! Replicator.Get(key, Replicator.ReadLocal, context.self.unsafeUpcast)
+ Behaviors.receiveMessagePartial {
+ case reply @ Replicator.GetSuccess(`key`) =>
+ val state = reply.get(key)
+ if (state.revision == revision) {
+ context.log.infoN(
+ "{}: Starting Sharded Daemon Process [{}] out of a total [{}] (revision [{}])",
+ sdpContext.name,
+ sdpContext.processNumber,
+ sdpContext.totalProcesses,
+ revision)
+ behaviorFactory(sdpContext).unsafeCast
+ } else {
+ context.log.warnN(
+ "{}: Tried to start an old revision of worker ([{}] but latest revision is [{}], started at {})",
+ sdpContext.name,
+ sdpContext.revision,
+ state.revision,
+ state.started)
+ Behaviors.stopped
+ }
+ case Replicator.NotFound(`key`) =>
+ if (revision == startRevision) {
+ // No state yet but initial revision, safe
+ context.log.infoN(
+ "{}: Starting Sharded Daemon Process [{}] out of a total [{}] (revision [{}] and no state found)",
+ sdpContext.name,
+ sdpContext.processNumber,
+ sdpContext.totalProcesses,
+ revision)
+ behaviorFactory(sdpContext).unsafeCast
+ } else {
+ context.log.error2(
+ "{}: Tried to start revision [{}] of worker but no ddata state found",
+ sdpContext.name,
+ sdpContext.revision)
+ Behaviors.stopped
+ }
+ }
+ }
+
+ }
+ }
+
+}
diff --git a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardingSerializer.scala b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardingSerializer.scala
index 7fd5eff092..a0158fe73d 100644
--- a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardingSerializer.scala
+++ b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardingSerializer.scala
@@ -15,10 +15,16 @@ package org.apache.pekko.cluster.sharding.typed.internal
import java.io.NotSerializableException
import java.nio.ByteBuffer
+import java.time.Instant
import org.apache.pekko
+import pekko.actor.typed.ActorRefResolver
+import pekko.actor.typed.scaladsl.adapter._
import pekko.annotation.InternalApi
+import pekko.cluster.sharding.typed.ChangeNumberOfProcesses
+import pekko.cluster.sharding.typed.GetNumberOfProcesses
import pekko.cluster.sharding.typed.ShardingEnvelope
+import pekko.cluster.sharding.typed.internal.ShardedDaemonProcessCoordinator.GetNumberOfProcessesReply
import pekko.cluster.sharding.typed.internal.protobuf.ShardingMessages
import pekko.protobufv3.internal.CodedOutputStream
import pekko.remote.serialization.WrappedPayloadSupport
@@ -35,12 +41,21 @@ import pekko.serialization.SerializerWithStringManifest
with BaseSerializer {
private val payloadSupport = new WrappedPayloadSupport(system)
+ private lazy val resolver = ActorRefResolver(system.toTyped)
private val ShardingEnvelopeManifest = "a"
+ private val DaemonProcessStateManifest = "b"
+ private val ChangeNumberOfProcessesManifest = "c"
+ private val GetNumberOfProcessesManifest = "d"
+ private val GetNumberOfProcessesReplyManifest = "e"
override def manifest(o: AnyRef): String = o match {
- case _: ShardingEnvelope[_] => ShardingEnvelopeManifest
- case _ =>
+ case _: ShardingEnvelope[_] => ShardingEnvelopeManifest
+ case _: ShardedDaemonProcessState => DaemonProcessStateManifest
+ case _: ChangeNumberOfProcesses => ChangeNumberOfProcessesManifest
+ case _: GetNumberOfProcesses => GetNumberOfProcessesManifest
+ case _: GetNumberOfProcessesReply => GetNumberOfProcessesReplyManifest
+ case _ =>
throw new IllegalArgumentException(s"Can't serialize object of type ${o.getClass} in [${getClass.getName}]")
}
@@ -51,6 +66,41 @@ import pekko.serialization.SerializerWithStringManifest
builder.setMessage(payloadSupport.payloadBuilder(env.message))
builder.build().toByteArray()
+ case state: ShardedDaemonProcessState =>
+ ShardingMessages.DaemonProcessScaleState
+ .newBuilder()
+ .setRevision(state.revision)
+ .setNumberOfProcesses(state.numberOfProcesses)
+ .setCompleted(state.completed)
+ .setStartedTimestampMillis(state.started.toEpochMilli)
+ .build()
+ .toByteArray()
+
+ case change: ChangeNumberOfProcesses =>
+ ShardingMessages.ChangeNumberOfProcesses
+ .newBuilder()
+ .setNewNumberOfProcesses(change.newNumberOfProcesses)
+ .setReplyTo(resolver.toSerializationFormat(change.replyTo))
+ .build()
+ .toByteArray()
+
+ case get: GetNumberOfProcesses =>
+ ShardingMessages.GetNumberOfProcesses
+ .newBuilder()
+ .setReplyTo(resolver.toSerializationFormat(get.replyTo))
+ .build()
+ .toByteArray()
+
+ case reply: GetNumberOfProcessesReply =>
+ ShardingMessages.GetNumberOfProcessesReply
+ .newBuilder()
+ .setRevision(reply.revision)
+ .setRescaleInProgress(reply.rescaleInProgress)
+ .setNumberOfProcesses(reply.numberOfProcesses)
+ .setStartedTimestampMillis(reply.started.toEpochMilli)
+ .build()
+ .toByteArray()
+
case _ =>
throw new IllegalArgumentException(s"Cannot serialize object of type [${o.getClass.getName}]")
}
@@ -61,6 +111,31 @@ import pekko.serialization.SerializerWithStringManifest
val entityId = env.getEntityId
val wrappedMsg = payloadSupport.deserializePayload(env.getMessage)
ShardingEnvelope(entityId, wrappedMsg)
+
+ case DaemonProcessStateManifest =>
+ val st = ShardingMessages.DaemonProcessScaleState.parseFrom(bytes)
+ ShardedDaemonProcessState(
+ st.getRevision,
+ st.getNumberOfProcesses,
+ st.getCompleted,
+ Instant.ofEpochMilli(st.getStartedTimestampMillis))
+
+ case ChangeNumberOfProcessesManifest =>
+ val change = ShardingMessages.ChangeNumberOfProcesses.parseFrom(bytes)
+ ChangeNumberOfProcesses(change.getNewNumberOfProcesses, resolver.resolveActorRef(change.getReplyTo))
+
+ case GetNumberOfProcessesManifest =>
+ val get = ShardingMessages.GetNumberOfProcesses.parseFrom(bytes)
+ GetNumberOfProcesses(resolver.resolveActorRef(get.getReplyTo))
+
+ case GetNumberOfProcessesReplyManifest =>
+ val reply = ShardingMessages.GetNumberOfProcessesReply.parseFrom(bytes)
+ GetNumberOfProcessesReply(
+ reply.getNumberOfProcesses,
+ Instant.ofEpochMilli(reply.getStartedTimestampMillis),
+ reply.getRescaleInProgress,
+ reply.getRevision)
+
case _ =>
throw new NotSerializableException(
s"Unimplemented deserialization of message with manifest [$manifest] in [${getClass.getName}]")
@@ -75,6 +150,50 @@ import pekko.serialization.SerializerWithStringManifest
val codedOutputStream = CodedOutputStream.newInstance(buf)
builder.build().writeTo(codedOutputStream)
codedOutputStream.flush()
+
+ case state: ShardedDaemonProcessState =>
+ val codedOutputStream = CodedOutputStream.newInstance(buf)
+ ShardingMessages.DaemonProcessScaleState
+ .newBuilder()
+ .setRevision(state.revision)
+ .setNumberOfProcesses(state.numberOfProcesses)
+ .setCompleted(state.completed)
+ .setStartedTimestampMillis(state.started.toEpochMilli)
+ .build()
+ .writeTo(codedOutputStream)
+ codedOutputStream.flush()
+
+ case change: ChangeNumberOfProcesses =>
+ val codedOutputStream = CodedOutputStream.newInstance(buf)
+ ShardingMessages.ChangeNumberOfProcesses
+ .newBuilder()
+ .setNewNumberOfProcesses(change.newNumberOfProcesses)
+ .setReplyTo(resolver.toSerializationFormat(change.replyTo))
+ .build()
+ .writeTo(codedOutputStream)
+ codedOutputStream.flush()
+
+ case get: GetNumberOfProcesses =>
+ val codedOutputStream = CodedOutputStream.newInstance(buf)
+ ShardingMessages.GetNumberOfProcesses
+ .newBuilder()
+ .setReplyTo(resolver.toSerializationFormat(get.replyTo))
+ .build()
+ .writeTo(codedOutputStream)
+ codedOutputStream.flush()
+
+ case reply: GetNumberOfProcessesReply =>
+ val codedOutputStream = CodedOutputStream.newInstance(buf)
+ ShardingMessages.GetNumberOfProcessesReply
+ .newBuilder()
+ .setRevision(reply.revision)
+ .setRescaleInProgress(reply.rescaleInProgress)
+ .setNumberOfProcesses(reply.numberOfProcesses)
+ .setStartedTimestampMillis(reply.started.toEpochMilli)
+ .build()
+ .writeTo(codedOutputStream)
+ codedOutputStream.flush()
+
case _ =>
throw new IllegalArgumentException(s"Cannot serialize object of type [${o.getClass.getName}]")
}
@@ -85,6 +204,31 @@ import pekko.serialization.SerializerWithStringManifest
val entityId = env.getEntityId
val wrappedMsg = payloadSupport.deserializePayload(env.getMessage)
ShardingEnvelope(entityId, wrappedMsg)
+
+ case DaemonProcessStateManifest =>
+ val st = ShardingMessages.DaemonProcessScaleState.parseFrom(buf)
+ ShardedDaemonProcessState(
+ st.getRevision,
+ st.getNumberOfProcesses,
+ st.getCompleted,
+ Instant.ofEpochMilli(st.getStartedTimestampMillis))
+
+ case ChangeNumberOfProcessesManifest =>
+ val change = ShardingMessages.ChangeNumberOfProcesses.parseFrom(buf)
+ ChangeNumberOfProcesses(change.getNewNumberOfProcesses, resolver.resolveActorRef(change.getReplyTo))
+
+ case GetNumberOfProcessesManifest =>
+ val get = ShardingMessages.GetNumberOfProcesses.parseFrom(buf)
+ GetNumberOfProcesses(resolver.resolveActorRef(get.getReplyTo))
+
+ case GetNumberOfProcessesReplyManifest =>
+ val reply = ShardingMessages.GetNumberOfProcessesReply.parseFrom(buf)
+ GetNumberOfProcessesReply(
+ reply.getNumberOfProcesses,
+ Instant.ofEpochMilli(reply.getStartedTimestampMillis),
+ reply.getRescaleInProgress,
+ reply.getRevision)
+
case _ =>
throw new NotSerializableException(
s"Unimplemented deserialization of message with manifest [$manifest] in [${getClass.getName}]")
diff --git a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/javadsl/ShardedDaemonProcess.scala b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/javadsl/ShardedDaemonProcess.scala
index 4644666ca3..408922f2ef 100644
--- a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/javadsl/ShardedDaemonProcess.scala
+++ b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/javadsl/ShardedDaemonProcess.scala
@@ -15,12 +15,17 @@ package org.apache.pekko.cluster.sharding.typed.javadsl
import java.util.Optional
import java.util.function.IntFunction
+import java.util.function.{ Function => JFunction }
import org.apache.pekko
+import pekko.actor.typed.ActorRef
import pekko.actor.typed.ActorSystem
import pekko.actor.typed.Behavior
+import pekko.annotation.ApiMayChange
import pekko.annotation.DoNotInherit
import pekko.cluster.sharding.ShardCoordinator.ShardAllocationStrategy
+import pekko.cluster.sharding.typed.ShardedDaemonProcessCommand
+import pekko.cluster.sharding.typed.ShardedDaemonProcessContext
import pekko.cluster.sharding.typed.ShardedDaemonProcessSettings
object ShardedDaemonProcess {
@@ -86,8 +91,8 @@ abstract class ShardedDaemonProcess {
/**
* Start a specific number of actors, each with a unique numeric id in the set, that is then kept alive in the cluster.
* @param behaviorFactory Given a unique id of `0` until `numberOfInstance` create the behavior for that actor.
- * @param stopMessage if defined sent to the actors when they need to stop because of a rebalance across the nodes of the cluster
- * or cluster shutdown.
+ * @param stopMessage if defined sent to the actors when they need to stop because of a rebalance across the nodes of the cluster,
+ * rescale or cluster shutdown.
* @param shardAllocationStrategy if defined used by entities to control the shard allocation
*/
def init[T](
@@ -99,4 +104,53 @@ abstract class ShardedDaemonProcess {
stopMessage: Optional[T],
shardAllocationStrategy: Optional[ShardAllocationStrategy]): Unit
+ /**
+ * Start a specific number of actors, each with a unique numeric id in the set, that is then kept alive in the cluster.
+ * The number of processing actors can be rescaled by interacting with the returned actor.
+ *
+ * @param behaviorFactory Given a unique id of `0` until `numberOfInstance` and total number of processes, create the behavior for that actor.
+ */
+ @ApiMayChange
+ def initWithContext[T](
+ messageClass: Class[T],
+ name: String,
+ initialNumberOfInstances: Int,
+ behaviorFactory: JFunction[ShardedDaemonProcessContext, Behavior[T]]): ActorRef[ShardedDaemonProcessCommand]
+
+ /**
+ * Start a specific number of actors, each with a unique numeric id in the set, that is then kept alive in the cluster.
+ * The number of processing actors can be rescaled by interacting with the returned actor.
+ *
+ * @param behaviorFactory Given a unique id of `0` until `numberOfInstance` and total number of processes, create the behavior for that actor.
+ * @param stopMessage Sent to the actors when they need to stop because of a rebalance across the nodes of the cluster
+ * or cluster shutdown.
+ */
+ @ApiMayChange
+ def initWithContext[T](
+ messageClass: Class[T],
+ name: String,
+ initialNumberOfInstances: Int,
+ behaviorFactory: JFunction[ShardedDaemonProcessContext, Behavior[T]],
+ settings: ShardedDaemonProcessSettings,
+ stopMessage: Optional[T]): ActorRef[ShardedDaemonProcessCommand]
+
+ /**
+ * Start a specific number of actors, each with a unique numeric id in the set, that is then kept alive in the cluster.
+ *
+ * @param behaviorFactory Given a unique sharded daemon process context containing the total number of workers and the id
+ * the specific worker being started, create the behavior for that actor.
+ * @param stopMessage If defined: sent to the actors when they need to stop because of a rebalance across the nodes of the cluster,
+ * rescale or cluster shutdown.
+ * @param shardAllocationStrategy If defined: used by entities to control the shard allocation
+ */
+ @ApiMayChange
+ def initWithContext[T](
+ messageClass: Class[T],
+ name: String,
+ numberOfInstances: Int,
+ behaviorFactory: JFunction[ShardedDaemonProcessContext, Behavior[T]],
+ settings: ShardedDaemonProcessSettings,
+ stopMessage: Optional[T],
+ shardAllocationStrategy: Optional[ShardAllocationStrategy]): ActorRef[ShardedDaemonProcessCommand]
+
}
diff --git a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/scaladsl/ShardedDaemonProcess.scala b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/scaladsl/ShardedDaemonProcess.scala
index 1e30bb4fb6..7155223f59 100644
--- a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/scaladsl/ShardedDaemonProcess.scala
+++ b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/scaladsl/ShardedDaemonProcess.scala
@@ -16,13 +16,17 @@ package org.apache.pekko.cluster.sharding.typed.scaladsl
import scala.reflect.ClassTag
import org.apache.pekko
+import pekko.actor.typed.ActorRef
import pekko.actor.typed.ActorSystem
import pekko.actor.typed.Behavior
import pekko.actor.typed.Extension
import pekko.actor.typed.ExtensionId
+import pekko.annotation.ApiMayChange
import pekko.annotation.DoNotInherit
import pekko.annotation.InternalApi
import pekko.cluster.sharding.ShardCoordinator.ShardAllocationStrategy
+import pekko.cluster.sharding.typed.ShardedDaemonProcessCommand
+import pekko.cluster.sharding.typed.ShardedDaemonProcessContext
import pekko.cluster.sharding.typed.ShardedDaemonProcessSettings
import pekko.cluster.sharding.typed.internal.ShardedDaemonProcessImpl
import pekko.cluster.sharding.typed.javadsl
@@ -93,6 +97,57 @@ trait ShardedDaemonProcess extends Extension { javadslSelf: javadsl.ShardedDaemo
stopMessage: Option[T],
shardAllocationStrategy: Option[ShardAllocationStrategy])(implicit classTag: ClassTag[T]): Unit
+ /**
+ * Start a specific number of actors, each with a unique numeric id in the set, that is then kept alive in the cluster.
+ * The number of processing actors can be rescaled by interacting with the returned actor.
+ *
+ * @param behaviorFactory Given a unique sharded daemon process context containing the total number of workers and the id
+ * the specific worker being started, create the behavior for that actor.
+ */
+ @ApiMayChange
+ def initWithContext[T](
+ name: String,
+ initialNumberOfInstances: Int,
+ behaviorFactory: ShardedDaemonProcessContext => Behavior[T])(
+ implicit classTag: ClassTag[T]): ActorRef[ShardedDaemonProcessCommand]
+
+ /**
+ * Start a specific number of actors, each with a unique numeric id in the set, that is then kept alive in the cluster.
+ * The number of processing actors can be rescaled by interacting with the returned actor.
+ *
+ * @param behaviorFactory Given a unique sharded daemon process context containing the total number of workers and the id
+ * the specific worker being started, create the behavior for that actor.
+ * @param stopMessage Sent to the actors when they need to stop because of a worker resize, re-balance across the
+ * nodes of the cluster or cluster shutdown.
+ */
+ @ApiMayChange
+ def initWithContext[T](
+ name: String,
+ initialNumberOfInstances: Int,
+ behaviorFactory: ShardedDaemonProcessContext => Behavior[T],
+ settings: ShardedDaemonProcessSettings,
+ stopMessage: T)(implicit classTag: ClassTag[T]): ActorRef[ShardedDaemonProcessCommand]
+
+ /**
+ * Start a specific number of actors, each with a unique numeric id in the set, that is then kept alive in the cluster.
+ * The number of processing actors can be rescaled by interacting with the returned actor.
+ *
+ * @param behaviorFactory Given a unique sharded daemon process context containing the total number of workers
+ * and the id the specific worker being started, create the behavior for that actor.
+ * @param stopMessage If defined, sent to the actors when they need to stop because of a worker resize,
+ * re-balance across the nodes of the cluster or cluster shutdown.
+ * @param shardAllocationStrategy If defined, used by entities to control the shard allocation.
+ */
+ @ApiMayChange
+ def initWithContext[T](
+ name: String,
+ numberOfInstances: Int,
+ behaviorFactory: ShardedDaemonProcessContext => Behavior[T],
+ settings: ShardedDaemonProcessSettings,
+ stopMessage: Option[T],
+ shardAllocationStrategy: Option[ShardAllocationStrategy])(
+ implicit classTag: ClassTag[T]): ActorRef[ShardedDaemonProcessCommand]
+
/**
* INTERNAL API
*/
diff --git a/cluster-sharding-typed/src/test/java/org/apache/pekko/cluster/sharding/typed/javadsl/ShardedDaemonProcessCompileOnlyTest.java b/cluster-sharding-typed/src/test/java/org/apache/pekko/cluster/sharding/typed/javadsl/ShardedDaemonProcessCompileOnlyTest.java
index d8c952d048..e759c2171a 100644
--- a/cluster-sharding-typed/src/test/java/org/apache/pekko/cluster/sharding/typed/javadsl/ShardedDaemonProcessCompileOnlyTest.java
+++ b/cluster-sharding-typed/src/test/java/org/apache/pekko/cluster/sharding/typed/javadsl/ShardedDaemonProcessCompileOnlyTest.java
@@ -13,12 +13,17 @@
package org.apache.pekko.cluster.sharding.typed.javadsl;
+import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
+import org.apache.pekko.actor.typed.ActorRef;
import org.apache.pekko.actor.typed.ActorSystem;
import org.apache.pekko.actor.typed.Behavior;
import org.apache.pekko.actor.typed.javadsl.Behaviors;
+import org.apache.pekko.cluster.sharding.ShardCoordinator;
+import org.apache.pekko.cluster.sharding.typed.ClusterShardingSettings;
+import org.apache.pekko.cluster.sharding.typed.ShardedDaemonProcessCommand;
import org.apache.pekko.cluster.sharding.typed.ShardedDaemonProcessSettings;
public class ShardedDaemonProcessCompileOnlyTest {
@@ -53,6 +58,62 @@ public class ShardedDaemonProcessCompileOnlyTest {
// #tag-processing
}
+ void coverAllFactories(ActorSystem<Void> system) {
+ ShardedDaemonProcess.get(system)
+ .init(TagProcessor.Command.class, "name", 8, id -> Behaviors.empty());
+
+ ShardedDaemonProcessSettings settings =
+ ShardedDaemonProcessSettings.create(system)
+ .withKeepAliveFromNumberOfNodes(7)
+ .withKeepAliveInterval(Duration.ofSeconds(2))
+ .withKeepAliveThrottleInterval(Duration.ofMillis(200))
+ .withShardingSettings(ClusterShardingSettings.create(system));
+
+ ShardedDaemonProcess.get(system)
+ .init(
+ TagProcessor.Command.class,
+ "name",
+ 8,
+ id -> Behaviors.empty(),
+ settings,
+ Optional.empty());
+
+ ShardedDaemonProcess.get(system)
+ .init(
+ TagProcessor.Command.class,
+ "name",
+ 8,
+ id -> Behaviors.empty(),
+ settings,
+ Optional.empty(),
+ Optional.of(new ShardCoordinator.LeastShardAllocationStrategy(10, 100)));
+
+ ActorRef<ShardedDaemonProcessCommand> control1 =
+ ShardedDaemonProcess.get(system)
+ .initWithContext(TagProcessor.Command.class, "name", 8, ctx -> Behaviors.empty());
+
+ ActorRef<ShardedDaemonProcessCommand> control2 =
+ ShardedDaemonProcess.get(system)
+ .initWithContext(
+ TagProcessor.Command.class,
+ "name",
+ 8,
+ ctx -> Behaviors.empty(),
+ settings,
+ Optional.empty());
+
+ ActorRef<ShardedDaemonProcessCommand> control3 =
+ ShardedDaemonProcess.get(system)
+ .initWithContext(
+ TagProcessor.Command.class,
+ "name",
+ 8,
+ ctx -> Behaviors.empty(),
+ settings,
+ Optional.empty(),
+ Optional.empty());
+ }
+
static class TagProcessor {
interface Command {}
diff --git a/cluster-sharding-typed/src/test/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessIdSpec.scala b/cluster-sharding-typed/src/test/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessIdSpec.scala
new file mode 100644
index 0000000000..f7f4d1fdd6
--- /dev/null
+++ b/cluster-sharding-typed/src/test/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessIdSpec.scala
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.cluster.sharding.typed.internal
+
+import org.apache.pekko.cluster.sharding.typed.internal.ShardedDaemonProcessId.DecodedId
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpecLike
+
+class ShardedDaemonProcessIdSpec extends AnyWordSpecLike with Matchers {
+
+ "The sharded daemon process ids" should {
+
+ "decode single int in id as id for non-rescaling" in {
+ // number of processes and revision not used so 0 in this case
+ ShardedDaemonProcessId.decodeEntityId("1", 8) should ===(DecodedId(0L, 8, 1))
+ }
+
+ "decode composed id as id for rescaling" in {
+ ShardedDaemonProcessId.decodeEntityId("1|4|3", 8) should ===(DecodedId(1L, 4, 3))
+ }
+
+ "decode single int in id as id for rescaling (to support rolling upgrade)" in {
+ ShardedDaemonProcessId.decodeEntityId("1", 8) should ===(DecodedId(0L, 8, 1))
+ }
+
+ "encode initial revision with old scheme (to support rolling upgrade)" in {
+ val id = DecodedId(0L, 8, 4)
+ id.encodeEntityId() should ===("4")
+ id.encodeEntityId() should ===("4")
+ }
+
+ "encode non 0 revisions with new scheme" in {
+ val id = DecodedId(1L, 8, 4)
+ id.encodeEntityId() should ===("1|8|4")
+ }
+
+ "spread out workers over shards for non-rescaling" in {
+ ShardedDaemonProcessId.allShardsFor(1, 8).size should ===(8)
+ }
+
+ "spread out workers over shards for rescaling" in {
+ ShardedDaemonProcessId.allShardsFor(1, 8).size should ===(8)
+ }
+
+ "extract new id as shard in message extractor" in {
+ val extractor = new ShardedDaemonProcessId.MessageExtractor[Any]()
+ extractor.shardId("1|8|4") should ===("4")
+ }
+
+ "extract old id as shard in message extractor (to support rolling upgrade)" in {
+ val extractor = new ShardedDaemonProcessId.MessageExtractor[Any]()
+ extractor.shardId("1") should ===("1")
+ }
+ }
+
+}
diff --git a/cluster-sharding-typed/src/test/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessStateSpec.scala b/cluster-sharding-typed/src/test/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessStateSpec.scala
new file mode 100644
index 0000000000..9b4f7cef43
--- /dev/null
+++ b/cluster-sharding-typed/src/test/scala/org/apache/pekko/cluster/sharding/typed/internal/ShardedDaemonProcessStateSpec.scala
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.cluster.sharding.typed.internal
+
+import java.time.Instant
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpecLike
+
+class ShardedDaemonProcessStateSpec extends AnyWordSpecLike with Matchers {
+
+ "The sharded daemon process state" should {
+
+ "merge highest revision" in {
+ val rev1 = ShardedDaemonProcessState(1L, 2, true, Instant.now())
+ val rev2 = ShardedDaemonProcessState(2L, 2, true, Instant.now())
+ rev1.merge(rev2) shouldBe theSameInstanceAs(rev2)
+ rev2.merge(rev1) shouldBe theSameInstanceAs(rev2)
+ rev1.merge(rev1) shouldBe theSameInstanceAs(rev1)
+ rev2.merge(rev2) shouldBe theSameInstanceAs(rev2)
+ }
+
+ "merge same revision, completion" in {
+ val incomplete = ShardedDaemonProcessState(2L, 2, false, Instant.now())
+ val complete = ShardedDaemonProcessState(2L, 2, true, Instant.now())
+ complete.merge(incomplete) shouldBe theSameInstanceAs(complete)
+ incomplete.merge(complete) shouldBe theSameInstanceAs(complete)
+ incomplete.merge(incomplete) shouldBe theSameInstanceAs(incomplete)
+ complete.merge(complete) shouldBe theSameInstanceAs(complete)
+ }
+
+ }
+
+}
diff --git a/cluster-sharding-typed/src/test/scala/org/apache/pekko/cluster/sharding/typed/scaladsl/ShardedDaemonProcessSpec.scala b/cluster-sharding-typed/src/test/scala/org/apache/pekko/cluster/sharding/typed/scaladsl/ShardedDaemonProcessSpec.scala
index 8251755462..529f04133b 100644
--- a/cluster-sharding-typed/src/test/scala/org/apache/pekko/cluster/sharding/typed/scaladsl/ShardedDaemonProcessSpec.scala
+++ b/cluster-sharding-typed/src/test/scala/org/apache/pekko/cluster/sharding/typed/scaladsl/ShardedDaemonProcessSpec.scala
@@ -22,8 +22,13 @@ import pekko.actor.typed.ActorRef
import pekko.actor.typed.Behavior
import pekko.actor.typed.scaladsl.Behaviors
import pekko.cluster.MemberStatus
+import pekko.cluster.ddata.typed.scaladsl.DistributedData
+import pekko.cluster.ddata.typed.scaladsl.Replicator
+import pekko.cluster.sharding.typed.ClusterShardingSettings
import pekko.cluster.sharding.typed.ShardedDaemonProcessSettings
-import pekko.cluster.sharding.typed.internal.ShardedDaemonProcessImpl.KeepAlivePinger
+import pekko.cluster.sharding.typed.internal.ShardedDaemonProcessCoordinator
+import pekko.cluster.sharding.typed.internal.ShardedDaemonProcessState
+import pekko.cluster.sharding.typed.internal.ShardedDaemonProcessStateKey
import pekko.cluster.typed.Cluster
import pekko.cluster.typed.Join
@@ -117,7 +122,7 @@ class ShardedDaemonProcessSpec
}
- "KeepAlivePinger" must {
+ "Coordinator" must {
"have a single node cluster running first" in {
val probe = createTestProbe()
Cluster(system).manager ! Join(Cluster(system).selfMember.address)
@@ -126,18 +131,61 @@ class ShardedDaemonProcessSpec
}, 3.seconds)
}
+ "send keep alive messages to original id scheme when revision is 0" in {
+ val shardingProbe = createTestProbe[Any]()
+ val settings = ShardedDaemonProcessSettings(system)
+ val shardingSettings = ClusterShardingSettings(system)
+ val pinger =
+ spawn(ShardedDaemonProcessCoordinator(settings, shardingSettings, 3, "throttle-a", shardingProbe.ref))
+ // note that StartEntity.apply is actually a ShardingEnvelope wrapping the StartEntity message
+ // See ShardedDaemonProcessId for details about entity id format
+ shardingProbe.expectMessage(StartEntity("0"))
+ shardingProbe.expectMessage(StartEntity("1"))
+ shardingProbe.expectMessage(StartEntity("2"))
+ shardingProbe.expectMessage(StartEntity("0"))
+
+ testKit.stop(pinger)
+ }
+
+ "send keep alive messages with new scheme when revision is > 0" in {
+ val name = "throttle-b"
+ val key = ShardedDaemonProcessStateKey(name)
+ val ddataProbe = createTestProbe[Replicator.UpdateResponse[ShardedDaemonProcessState]]()
+ DistributedData(system).replicator.tell(
+ Replicator.Update(key, ShardedDaemonProcessState.initialState(2), Replicator.WriteLocal, ddataProbe.ref)(
+ state => state.startScalingTo(3).completeScaling()))
+ ddataProbe.expectMessageType[Replicator.UpdateSuccess[ShardedDaemonProcessState]]
+
+ val shardingProbe = createTestProbe[Any]()
+ val settings = ShardedDaemonProcessSettings(system)
+ val shardingSettings = ClusterShardingSettings(system)
+ val pinger =
+ spawn(ShardedDaemonProcessCoordinator(settings, shardingSettings, 3, name, shardingProbe.ref))
+ // note that StartEntity.apply is actually a ShardingEnvelope wrapping the StartEntity message
+ // See ShardedDaemonProcessId for details about entity id format
+ shardingProbe.expectMessage(StartEntity("1|3|0"))
+ shardingProbe.expectMessage(StartEntity("1|3|1"))
+ shardingProbe.expectMessage(StartEntity("1|3|2"))
+ shardingProbe.expectMessage(StartEntity("1|3|0"))
+
+ testKit.stop(pinger)
+ }
+
"throttle keep alive messages" in {
val shardingProbe = createTestProbe[Any]()
val settings = ShardedDaemonProcessSettings(system).withKeepAliveThrottleInterval(1.second)
- val pinger = spawn(KeepAlivePinger(settings, "throttle", Set("1", "2", "3"), shardingProbe.ref))
+ val shardingSettings = ClusterShardingSettings(system)
+ val pinger =
+ spawn(ShardedDaemonProcessCoordinator(settings, shardingSettings, 3, "throttle-c", shardingProbe.ref))
// note that StartEntity.apply is actually a ShardingEnvelope wrapping the StartEntity message
+ // See ShardedDaemonProcessId for details about entity id format
+ shardingProbe.expectMessage(StartEntity("0"))
+ shardingProbe.expectNoMessage(100.millis)
shardingProbe.expectMessage(StartEntity("1"))
shardingProbe.expectNoMessage(100.millis)
shardingProbe.expectMessage(StartEntity("2"))
shardingProbe.expectNoMessage(100.millis)
- shardingProbe.expectMessage(StartEntity("3"))
- shardingProbe.expectNoMessage(100.millis)
- shardingProbe.expectMessage(StartEntity("1"))
+ shardingProbe.expectMessage(StartEntity("0"))
testKit.stop(pinger)
}
diff --git a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala
index c29a8624da..1729d14e11 100644
--- a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala
+++ b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala
@@ -426,6 +426,11 @@ object ShardCoordinator {
*/
@SerialVersionUID(1L) final case class RegionStopped(shardRegion: ActorRef) extends CoordinatorCommand
+ /**
+ * Stop all the listed shards, sender will get a ShardStopped ack for each shard once stopped
+ */
+ final case class StopShards(shards: Set[ShardId]) extends CoordinatorCommand
+
/**
* `ShardRegion` requests full handoff to be able to shutdown gracefully.
*/
@@ -690,6 +695,7 @@ abstract class ShardCoordinator(
var rebalanceInProgress = Map.empty[ShardId, Set[ActorRef]]
var rebalanceWorkers: Set[ActorRef] = Set.empty
var unAckedHostShards = Map.empty[ShardId, Cancellable]
+ var waitingForShardsToStop: Map[ShardId, Set[(ActorRef, java.util.UUID)]] = Map.empty
// regions that have requested handoff, for graceful shutdown
var gracefulShutdownInProgress = Set.empty[ActorRef]
var waitingForLocalRegionToTerminate = false
@@ -844,6 +850,14 @@ abstract class ShardCoordinator(
if (ok) {
log.debug("{}: Shard [{}] deallocation completed successfully.", typeName, shard)
+ // ack to external trigger of rebalance/stop
+ waitingForShardsToStop.get(shard) match {
+ case Some(waiting) =>
+ waiting.foreach { case (replyTo, _) => replyTo ! ShardStopped(shard) }
+ waitingForShardsToStop -= shard
+ case None =>
+ }
+
// The shard could have been removed by ShardRegionTerminated
if (state.shards.contains(shard)) {
update(ShardHomeDeallocated(shard)) { evt =>
@@ -1092,6 +1106,15 @@ abstract class ShardCoordinator(
case DelayedShardRegionTerminated(ref) =>
regionTerminated(ref)
+
+ case StopShards(shards) =>
+ val requestId = java.util.UUID.randomUUID()
+ shards.foreach { shard =>
+ waitingForShardsToStop = waitingForShardsToStop.updated(
+ shard,
+ waitingForShardsToStop.getOrElse(shard, Set.empty) + ((sender(), requestId)))
+ }
+ shutdownShards(self, shards.filter(state.shards.contains))
}
def update[E <: DomainEvent](evt: E)(f: E => Unit): Unit
@@ -1291,7 +1314,10 @@ abstract class ShardCoordinator(
def shutdownShards(shuttingDownRegion: ActorRef, shards: Set[ShardId]): Unit = {
if ((log: BusLogging).isInfoEnabled && (shards.nonEmpty)) {
- log.info("{}: Starting shutting down shards [{}] due to region shutting down.", typeName, shards.mkString(","))
+ log.info(
+ "{}: Starting shutting down shards [{}] due to region shutting down or explicit stopping of shards.",
+ typeName,
+ shards.mkString(","))
}
shards.foreach { shard =>
startShardRebalanceIfNeeded(shard, shuttingDownRegion, handOffTimeout, isRebalance = false)
diff --git a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardRegion.scala b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardRegion.scala
index eb5f263aec..b3a5cce4de 100644
--- a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardRegion.scala
+++ b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardRegion.scala
@@ -765,6 +765,7 @@ private[pekko] class ShardRegion(
case msg: RestartShard => deliverMessage(msg, sender())
case msg: StartEntity => deliverStartEntity(msg, sender())
case msg: SetActiveEntityLimit => deliverToAllShards(msg, sender())
+ case cmd: CoordinatorCommand => deliverCoordinatorCommand(cmd, sender())
case msg if extractEntityId.isDefinedAt(msg) => deliverMessage(msg, sender())
case unknownMsg =>
log.warning("{}: Message does not have an extractor defined in shard so it was ignored: {}", typeName, unknownMsg)
@@ -958,6 +959,10 @@ private[pekko] class ShardRegion(
case _ => unhandled(cmd)
}
+ def deliverCoordinatorCommand(cmd: CoordinatorCommand, snd: ActorRef): Unit = {
+ coordinator.foreach(_.tell(cmd, snd))
+ }
+
def receiveQuery(query: ShardRegionQuery): Unit = query match {
case GetCurrentRegions =>
coordinator match {
diff --git a/distributed-data/src/main/java/org/apache/pekko/cluster/ddata/protobuf/msg/ReplicatorMessages.java b/distributed-data/src/main/java/org/apache/pekko/cluster/ddata/protobuf/msg/ReplicatorMessages.java
index 8fc406ea5c..b8fd9c62d8 100644
--- a/distributed-data/src/main/java/org/apache/pekko/cluster/ddata/protobuf/msg/ReplicatorMessages.java
+++ b/distributed-data/src/main/java/org/apache/pekko/cluster/ddata/protobuf/msg/ReplicatorMessages.java
@@ -11699,7 +11699,7 @@ public final class ReplicatorMessages extends org.apache.pekko.protobufv3.intern
}
public static final int USEDTIMESTAMP_FIELD_NUMBER = 3;
- private long usedTimestamp_;
+ private long usedTimestamp_ = 0L;
/**
* <code>optional sint64 usedTimestamp = 3;</code>
* @return Whether the usedTimestamp field is set.
@@ -12212,7 +12212,7 @@ public final class ReplicatorMessages extends org.apache.pekko.protobufv3.intern
return this;
}
- private long usedTimestamp_;
+ private long usedTimestamp_ ;
/**
* <code>optional sint64 usedTimestamp = 3;</code>
* @return Whether the usedTimestamp field is set.
@@ -12235,6 +12235,7 @@ public final class ReplicatorMessages extends org.apache.pekko.protobufv3.intern
* @return This builder for chaining.
*/
public Builder setUsedTimestamp(long value) {
+
usedTimestamp_ = value;
bitField0_ |= 0x00000004;
onChanged();
@@ -13619,7 +13620,7 @@ public final class ReplicatorMessages extends org.apache.pekko.protobufv3.intern
}
public static final int USEDTIMESTAMP_FIELD_NUMBER = 3;
- private long usedTimestamp_;
+ private long usedTimestamp_ = 0L;
/**
* <code>optional sint64 usedTimestamp = 3;</code>
* @return Whether the usedTimestamp field is set.
@@ -14234,9 +14235,7 @@ public final class ReplicatorMessages extends org.apache.pekko.protobufv3.intern
return envelopeBuilder_;
}
- // @@protoc_insertion_point(builder_scope:org.apache.pekko.cluster.ddata.Gossip.Entry)
-
- private long usedTimestamp_;
+ private long usedTimestamp_ ;
/**
* <code>optional sint64 usedTimestamp = 3;</code>
* @return Whether the usedTimestamp field is set.
@@ -14259,6 +14258,7 @@ public final class ReplicatorMessages extends org.apache.pekko.protobufv3.intern
* @return This builder for chaining.
*/
public Builder setUsedTimestamp(long value) {
+
usedTimestamp_ = value;
bitField0_ |= 0x00000004;
onChanged();
@@ -14274,6 +14274,8 @@ public final class ReplicatorMessages extends org.apache.pekko.protobufv3.intern
onChanged();
return this;
}
+
+ // @@protoc_insertion_point(builder_scope:org.apache.pekko.cluster.ddata.Gossip.Entry)
}
// @@protoc_insertion_point(class_scope:org.apache.pekko.cluster.ddata.Gossip.Entry)
@@ -22729,41 +22731,42 @@ public final class ReplicatorMessages extends org.apache.pekko.protobufv3.intern
".apache.pekko.cluster.ddata.UniqueAddres" +
"s\022\021\n\tperformed\030\003 \002(\010\0225\n\004seen\030\004 \003(\0132\'.org" +
".apache.pekko.cluster.ddata.Address\022\024\n\014o" +
- "bsoleteTime\030\005 \001(\022\"\273\001\n\006Status\022\r\n\005chunk\030\001 " +
+ "bsoleteTime\030\005 \001(\022\"\322\001\n\006Status\022\r\n\005chunk\030\001 " +
"\002(\r\022\021\n\ttotChunks\030\002 \002(\r\022=\n\007entries\030\003 \003(\0132" +
",.org.apache.pekko.cluster.ddata.Status." +
"Entry\022\023\n\013toSystemUid\030\004 \001(\020\022\025\n\rfromSystem" +
- "Uid\030\005 \001(\020\032$\n\005Entry\022\013\n\003key\030\001 \002(\t\022\016\n\006diges" +
- "t\030\002 \002(\014\"\333\001\n\006Gossip\022\020\n\010sendBack\030\001 \002(\010\022=\n\007" +
- "entries\030\002 \003(\0132,.org.apache.pekko.cluster" +
- ".ddata.Gossip.Entry\022\023\n\013toSystemUid\030\003 \001(\020" +
- "\022\025\n\rfromSystemUid\030\004 \001(\020\032T\n\005Entry\022\013\n\003key\030" +
- "\001 \002(\t\022>\n\010envelope\030\002 \002(\0132,.org.apache.pek" +
- "ko.cluster.ddata.DataEnvelope\"\245\002\n\020DeltaP" +
- "ropagation\022?\n\010fromNode\030\001 \002(\0132-.org.apach" +
- "e.pekko.cluster.ddata.UniqueAddress\022G\n\007e" +
- "ntries\030\002 \003(\01326.org.apache.pekko.cluster." +
- "ddata.DeltaPropagation.Entry\022\r\n\005reply\030\003 " +
- "\001(\010\032x\n\005Entry\022\013\n\003key\030\001 \002(\t\022>\n\010envelope\030\002 " +
- "\002(\0132,.org.apache.pekko.cluster.ddata.Dat" +
- "aEnvelope\022\021\n\tfromSeqNr\030\003 \002(\003\022\017\n\007toSeqNr\030" +
- "\004 \001(\003\"d\n\rUniqueAddress\0228\n\007address\030\001 \002(\0132" +
- "\'.org.apache.pekko.cluster.ddata.Address" +
- "\022\013\n\003uid\030\002 \002(\017\022\014\n\004uid2\030\003 \001(\017\")\n\007Address\022\020" +
- "\n\010hostname\030\001 \002(\t\022\014\n\004port\030\002 \002(\r\"\254\001\n\rVersi" +
- "onVector\022D\n\007entries\030\001 \003(\01323.org.apache.p" +
- "ekko.cluster.ddata.VersionVector.Entry\032U" +
- "\n\005Entry\022;\n\004node\030\001 \002(\0132-.org.apache.pekko" +
- ".cluster.ddata.UniqueAddress\022\017\n\007version\030" +
- "\002 \002(\003\"V\n\014OtherMessage\022\027\n\017enclosedMessage" +
- "\030\001 \002(\014\022\024\n\014serializerId\030\002 \002(\005\022\027\n\017messageM" +
- "anifest\030\004 \001(\014\"\036\n\nStringGSet\022\020\n\010elements\030" +
- "\001 \003(\t\"\235\001\n\023DurableDataEnvelope\022:\n\004data\030\001 " +
- "\002(\0132,.org.apache.pekko.cluster.ddata.Oth" +
- "erMessage\022J\n\007pruning\030\002 \003(\01329.org.apache." +
- "pekko.cluster.ddata.DataEnvelope.Pruning" +
- "EntryB/\n+org.apache.pekko.cluster.ddata." +
- "protobuf.msgH\001"
+ "Uid\030\005 \001(\020\032;\n\005Entry\022\013\n\003key\030\001 \002(\t\022\016\n\006diges" +
+ "t\030\002 \002(\014\022\025\n\rusedTimestamp\030\003 \001(\022\"\362\001\n\006Gossi" +
+ "p\022\020\n\010sendBack\030\001 \002(\010\022=\n\007entries\030\002 \003(\0132,.o" +
+ "rg.apache.pekko.cluster.ddata.Gossip.Ent" +
+ "ry\022\023\n\013toSystemUid\030\003 \001(\020\022\025\n\rfromSystemUid" +
+ "\030\004 \001(\020\032k\n\005Entry\022\013\n\003key\030\001 \002(\t\022>\n\010envelope" +
+ "\030\002 \002(\0132,.org.apache.pekko.cluster.ddata." +
+ "DataEnvelope\022\025\n\rusedTimestamp\030\003 \001(\022\"\245\002\n\020" +
+ "DeltaPropagation\022?\n\010fromNode\030\001 \002(\0132-.org" +
+ ".apache.pekko.cluster.ddata.UniqueAddres" +
+ "s\022G\n\007entries\030\002 \003(\01326.org.apache.pekko.cl" +
+ "uster.ddata.DeltaPropagation.Entry\022\r\n\005re" +
+ "ply\030\003 \001(\010\032x\n\005Entry\022\013\n\003key\030\001 \002(\t\022>\n\010envel" +
+ "ope\030\002 \002(\0132,.org.apache.pekko.cluster.dda" +
+ "ta.DataEnvelope\022\021\n\tfromSeqNr\030\003 \002(\003\022\017\n\007to" +
+ "SeqNr\030\004 \001(\003\"d\n\rUniqueAddress\0228\n\007address\030" +
+ "\001 \002(\0132\'.org.apache.pekko.cluster.ddata.A" +
+ "ddress\022\013\n\003uid\030\002 \002(\017\022\014\n\004uid2\030\003 \001(\017\")\n\007Add" +
+ "ress\022\020\n\010hostname\030\001 \002(\t\022\014\n\004port\030\002 \002(\r\"\254\001\n" +
+ "\rVersionVector\022D\n\007entries\030\001 \003(\01323.org.ap" +
+ "ache.pekko.cluster.ddata.VersionVector.E" +
+ "ntry\032U\n\005Entry\022;\n\004node\030\001 \002(\0132-.org.apache" +
+ ".pekko.cluster.ddata.UniqueAddress\022\017\n\007ve" +
+ "rsion\030\002 \002(\003\"V\n\014OtherMessage\022\027\n\017enclosedM" +
+ "essage\030\001 \002(\014\022\024\n\014serializerId\030\002 \002(\005\022\027\n\017me" +
+ "ssageManifest\030\004 \001(\014\"\036\n\nStringGSet\022\020\n\010ele" +
+ "ments\030\001 \003(\t\"\235\001\n\023DurableDataEnvelope\022:\n\004d" +
+ "ata\030\001 \002(\0132,.org.apache.pekko.cluster.dda" +
+ "ta.OtherMessage\022J\n\007pruning\030\002 \003(\01329.org.a" +
+ "pache.pekko.cluster.ddata.DataEnvelope.P" +
+ "runingEntryB/\n+org.apache.pekko.cluster." +
+ "ddata.protobuf.msgH\001"
};
descriptor = org.apache.pekko.protobufv3.internal.Descriptors.FileDescriptor
.internalBuildGeneratedFileFrom(descriptorData,
@@ -22858,7 +22861,7 @@ public final class ReplicatorMessages extends org.apache.pekko.protobufv3.intern
internal_static_org_apache_pekko_cluster_ddata_Status_Entry_fieldAccessorTable = new
org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable(
internal_static_org_apache_pekko_cluster_ddata_Status_Entry_descriptor,
- new java.lang.String[] { "Key", "Digest", });
+ new java.lang.String[] { "Key", "Digest", "UsedTimestamp", });
internal_static_org_apache_pekko_cluster_ddata_Gossip_descriptor =
getDescriptor().getMessageType(13);
internal_static_org_apache_pekko_cluster_ddata_Gossip_fieldAccessorTable = new
@@ -22870,7 +22873,7 @@ public final class ReplicatorMessages extends org.apache.pekko.protobufv3.intern
internal_static_org_apache_pekko_cluster_ddata_Gossip_Entry_fieldAccessorTable = new
org.apache.pekko.protobufv3.internal.GeneratedMessage.FieldAccessorTable(
internal_static_org_apache_pekko_cluster_ddata_Gossip_Entry_descriptor,
- new java.lang.String[] { "Key", "Envelope", });
+ new java.lang.String[] { "Key", "Envelope", "UsedTimestamp", });
internal_static_org_apache_pekko_cluster_ddata_DeltaPropagation_descriptor =
getDescriptor().getMessageType(14);
internal_static_org_apache_pekko_cluster_ddata_DeltaPropagation_fieldAccessorTable = new
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org