feat(flagd): Adjust to disable-sync-metadata toggle in flagd

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/FlagStore.java

@@ -2,12 +2,12 @@
 
 import static dev.openfeature.contrib.providers.flagd.resolver.common.Convert.convertProtobufMapToStructure;
 
+import com.google.protobuf.Struct;
 import dev.openfeature.contrib.providers.flagd.resolver.process.model.FeatureFlag;
 import dev.openfeature.contrib.providers.flagd.resolver.process.model.FlagParser;
 import dev.openfeature.contrib.providers.flagd.resolver.process.model.ParsingResult;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.QueuePayload;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.QueueSource;
-import dev.openfeature.flagd.grpc.sync.Sync.GetMetadataResponse;
 import dev.openfeature.sdk.ImmutableStructure;
 import dev.openfeature.sdk.Structure;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
@@ -114,7 +114,7 @@ private void streamerListener(final QueueSource connector) throws InterruptedExc
                         Map<String, FeatureFlag> flagMap = parsingResult.getFlags();
                         Map<String, Object> flagSetMetadataMap = parsingResult.getFlagSetMetadata();
 
-                        Structure metadata = parseSyncMetadata(payload.getMetadataResponse());
+                        Structure metadata = parseSyncContextOrMetadata(payload.getSyncContext());
                         writeLock.lock();
                         try {
                             changedFlagsKeys = getChangedFlagsKeys(flagMap);
@@ -150,9 +150,9 @@ private void streamerListener(final QueueSource connector) throws InterruptedExc
         log.info("Shutting down store stream listener");
     }
 
-    private Structure parseSyncMetadata(GetMetadataResponse metadataResponse) {
+    private Structure parseSyncContextOrMetadata(Struct syncContext) {
         try {
-            return convertProtobufMapToStructure(metadataResponse.getMetadata().getFieldsMap());
+            return convertProtobufMapToStructure(syncContext.getFieldsMap());
         } catch (Exception exception) {
             log.error("Failed to parse metadataResponse, provider metadata may not be up-to-date");
         }

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/QueuePayload.java

@@ -1,6 +1,6 @@
 package dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector;
 
-import dev.openfeature.flagd.grpc.sync.Sync.GetMetadataResponse;
+import com.google.protobuf.Struct;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 
@@ -10,9 +10,9 @@
 public class QueuePayload {
     private final QueuePayloadType type;
     private final String flagData;
-    private final GetMetadataResponse metadataResponse;
+    private final Struct syncContext;
 
     public QueuePayload(QueuePayloadType type, String flagData) {
-        this(type, flagData, GetMetadataResponse.getDefaultInstance());
+        this(type, flagData, null);
     }
 }

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/sync/SyncStreamQueueSource.java

@@ -1,5 +1,6 @@
 package dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.sync;
 
+import com.google.protobuf.Struct;
 import dev.openfeature.contrib.providers.flagd.FlagdOptions;
 import dev.openfeature.contrib.providers.flagd.resolver.common.ChannelConnector;
 import dev.openfeature.contrib.providers.flagd.resolver.common.FlagdProviderEvent;
@@ -123,7 +124,7 @@ private void observeSyncStream() throws InterruptedException {
 
             log.debug("Initializing sync stream request");
             final GetMetadataRequest.Builder metadataRequest = GetMetadataRequest.newBuilder();
-            GetMetadataResponse metadataResponse = GetMetadataResponse.getDefaultInstance();
+            GetMetadataResponse metadataResponse = null;
 
             // create a context which exists to track and cancel the stream
             try (CancellableContext context = Context.current().withCancellation()) {
@@ -162,8 +163,7 @@ private void observeSyncStream() throws InterruptedException {
                         log.debug("Exception in stream RPC, streamException {}, will restart", streamException);
                         if (!outgoingQueue.offer(new QueuePayload(
                                 QueuePayloadType.ERROR,
-                                String.format("Error from stream: %s", streamException.getMessage()),
-                                metadataResponse))) {
+                                String.format("Error from stream: %s", streamException.getMessage())))) {
                             log.error("Failed to convey ERROR status, queue is full");
                         }
                         break;
@@ -173,7 +173,14 @@ private void observeSyncStream() throws InterruptedException {
                     final String data = flagsResponse.getFlagConfiguration();
                     log.debug("Got stream response: {}", data);
 
-                    if (!outgoingQueue.offer(new QueuePayload(QueuePayloadType.DATA, data, metadataResponse))) {
+                    Struct syncContext = null;
+                    if (flagsResponse.hasSyncContext()) {
+                        syncContext = flagsResponse.getSyncContext();
+                    } else if (metadataResponse != null) {
+                        syncContext = metadataResponse.getMetadata();
+                    }
+
+                    if (!outgoingQueue.offer(new QueuePayload(QueuePayloadType.DATA, data, syncContext))) {
                         log.error("Stream writing failed");
                     }
                 }

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/e2e/steps/ProviderSteps.java

@@ -120,7 +120,11 @@ public void setupProvider(String providerType) throws InterruptedException {
                     state.builder.port(container.getPort(State.resolverType));
                 }
                 break;
-            default:
+            case "syncpayload":
+                flagdConfig = "sync-payload";
+                state.builder.port(container.getPort(State.resolverType));
+                break;
+            case "stable":
                 this.state.providerType = ProviderType.DEFAULT;
                 if (State.resolverType == Config.Resolver.FILE) {
 
@@ -134,6 +138,8 @@ public void setupProvider(String providerType) throws InterruptedException {
                     state.builder.port(container.getPort(State.resolverType));
                 }
                 break;
+            default:
+                throw new IllegalStateException();
         }
         when().post("http://" + container.getLaunchpadUrl() + "/start?config={config}", flagdConfig)
                 .then()

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/FlagStoreTest.java

@@ -11,7 +11,6 @@
 import dev.openfeature.contrib.providers.flagd.resolver.process.model.FlagParser;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.QueuePayload;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.QueuePayloadType;
-import dev.openfeature.flagd.grpc.sync.Sync.GetMetadataResponse;
 import java.time.Duration;
 import java.util.HashSet;
 import java.util.Map;
@@ -35,10 +34,7 @@ void connectorHandling() throws Exception {
 
         // OK for simple flag
         assertTimeoutPreemptively(Duration.ofMillis(maxDelay), () -> {
-            payload.offer(new QueuePayload(
-                    QueuePayloadType.DATA,
-                    getFlagsFromResource(VALID_SIMPLE),
-                    GetMetadataResponse.getDefaultInstance()));
+            payload.offer(new QueuePayload(QueuePayloadType.DATA, getFlagsFromResource(VALID_SIMPLE)));
         });
 
         assertTimeoutPreemptively(Duration.ofMillis(maxDelay), () -> {
@@ -47,10 +43,7 @@ void connectorHandling() throws Exception {
 
         // STALE for invalid flag
         assertTimeoutPreemptively(Duration.ofMillis(maxDelay), () -> {
-            payload.offer(new QueuePayload(
-                    QueuePayloadType.DATA,
-                    getFlagsFromResource(INVALID_FLAG),
-                    GetMetadataResponse.getDefaultInstance()));
+            payload.offer(new QueuePayload(QueuePayloadType.DATA, getFlagsFromResource(INVALID_FLAG)));
         });
 
         assertTimeoutPreemptively(Duration.ofMillis(maxDelay), () -> {
@@ -59,8 +52,7 @@ void connectorHandling() throws Exception {
 
         // OK again for next payload
         assertTimeoutPreemptively(Duration.ofMillis(maxDelay), () -> {
-            payload.offer(new QueuePayload(
-                    QueuePayloadType.DATA, getFlagsFromResource(VALID_LONG), GetMetadataResponse.getDefaultInstance()));
+            payload.offer(new QueuePayload(QueuePayloadType.DATA, getFlagsFromResource(VALID_LONG)));
         });
 
         assertTimeoutPreemptively(Duration.ofMillis(maxDelay), () -> {
@@ -69,7 +61,7 @@ void connectorHandling() throws Exception {
 
         // ERROR is propagated correctly
         assertTimeoutPreemptively(Duration.ofMillis(maxDelay), () -> {
-            payload.offer(new QueuePayload(QueuePayloadType.ERROR, null, GetMetadataResponse.getDefaultInstance()));
+            payload.offer(new QueuePayload(QueuePayloadType.ERROR, null));
         });
 
         assertTimeoutPreemptively(Duration.ofMillis(maxDelay), () -> {
@@ -93,10 +85,7 @@ public void changedFlags() throws Exception {
         final BlockingQueue<StorageStateChange> storageStateDTOS = store.getStateQueue();
 
         assertTimeoutPreemptively(Duration.ofMillis(maxDelay), () -> {
-            payload.offer(new QueuePayload(
-                    QueuePayloadType.DATA,
-                    getFlagsFromResource(VALID_SIMPLE),
-                    GetMetadataResponse.getDefaultInstance()));
+            payload.offer(new QueuePayload(QueuePayloadType.DATA, getFlagsFromResource(VALID_SIMPLE)));
         });
         // flags changed for first time
         assertEquals(
@@ -105,8 +94,7 @@ public void changedFlags() throws Exception {
                 storageStateDTOS.take().getChangedFlagsKeys());
 
         assertTimeoutPreemptively(Duration.ofMillis(maxDelay), () -> {
-            payload.offer(new QueuePayload(
-                    QueuePayloadType.DATA, getFlagsFromResource(VALID_LONG), GetMetadataResponse.getDefaultInstance()));
+            payload.offer(new QueuePayload(QueuePayloadType.DATA, getFlagsFromResource(VALID_LONG)));
         });
         Map<String, FeatureFlag> expectedChangedFlags =
                 FlagParser.parseString(getFlagsFromResource(VALID_LONG), true).getFlags();

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/MockConnector.java

@@ -3,7 +3,6 @@
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.QueuePayload;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.QueuePayloadType;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.QueueSource;
-import dev.openfeature.flagd.grpc.sync.Sync.GetMetadataResponse;
 import java.util.concurrent.BlockingQueue;
 import lombok.extern.slf4j.Slf4j;
 
@@ -26,8 +25,7 @@ public BlockingQueue<QueuePayload> getStreamQueue() {
 
     public void shutdown() {
         // Emit error mocking closed connection scenario
-        if (!mockQueue.offer(new QueuePayload(
-                QueuePayloadType.ERROR, "shutdown invoked", GetMetadataResponse.getDefaultInstance()))) {
+        if (!mockQueue.offer(new QueuePayload(QueuePayloadType.ERROR, "shutdown invoked"))) {
             log.warn("Failed to offer shutdown status");
         }
     }