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 syncContext = parseSyncContext(payload.getSyncContext());
                         writeLock.lock();
                         try {
                             changedFlagsKeys = getChangedFlagsKeys(flagMap);
@@ -126,7 +126,7 @@ private void streamerListener(final QueueSource connector) throws InterruptedExc
                             writeLock.unlock();
                         }
                         if (!stateBlockingQueue.offer(
-                                new StorageStateChange(StorageState.OK, changedFlagsKeys, metadata))) {
+                                new StorageStateChange(StorageState.OK, changedFlagsKeys, syncContext))) {
                             log.warn("Failed to convey OK status, queue is full");
                         }
                     } catch (Throwable e) {
@@ -150,11 +150,13 @@ private void streamerListener(final QueueSource connector) throws InterruptedExc
         log.info("Shutting down store stream listener");
     }
 
-    private Structure parseSyncMetadata(GetMetadataResponse metadataResponse) {
-        try {
-            return convertProtobufMapToStructure(metadataResponse.getMetadata().getFieldsMap());
-        } catch (Exception exception) {
-            log.error("Failed to parse metadataResponse, provider metadata may not be up-to-date");
+    private Structure parseSyncContext(Struct syncContext) {
+        if (syncContext != null) {
+            try {
+                return convertProtobufMapToStructure(syncContext.getFieldsMap());
+            } catch (Exception exception) {
+                log.error("Failed to parse metadataResponse, provider metadata may not be up-to-date");
+            }
         }
         return new ImmutableStructure();
     }

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");
         }
     }

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/sync/SyncStreamQueueSourceTest.java

@@ -1,6 +1,7 @@
 package dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.sync;
 
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.ArgumentMatchers.any;
@@ -12,6 +13,7 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+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.QueueingStreamObserver;
@@ -73,6 +75,7 @@ void onNextEnqueuesDataPayload() throws Exception {
         BlockingQueue<QueuePayload> streamQueue = connector.getStreamQueue();
         QueuePayload payload = streamQueue.poll(1000, TimeUnit.MILLISECONDS);
         assertNotNull(payload);
+        assertNotNull(payload.getSyncContext());
         assertEquals(QueuePayloadType.DATA, payload.getType());
         // should NOT have restarted the stream (1 call)
         verify(stub, times(1)).syncFlags(any(), any());
@@ -94,13 +97,38 @@ void onNextEnqueuesDataPayloadMetadataDisabled() throws Exception {
         BlockingQueue<QueuePayload> streamQueue = connector.getStreamQueue();
         QueuePayload payload = streamQueue.poll(1000, TimeUnit.MILLISECONDS);
         assertNotNull(payload);
+        assertNull(payload.getSyncContext());
         assertEquals(QueuePayloadType.DATA, payload.getType());
         // should NOT have restarted the stream (1 call)
         verify(stub, times(1)).syncFlags(any(), any());
         // should NOT have called getMetadata
         verify(blockingStub, times(0)).getMetadata(any());
     }
 
+    @Test
+    void onNextEnqueuesDataPayloadWithSyncContext() throws Exception {
+        // disable GetMetadata call
+        SyncStreamQueueSource connector =
+                new SyncStreamQueueSource(FlagdOptions.builder().build(), mockConnector, stub, blockingStub);
+        latch = new CountDownLatch(1);
+        connector.init();
+        latch.await();
+
+        // fire onNext (data) event
+        Struct syncContext = Struct.newBuilder().build();
+        observer.onNext(
+                SyncFlagsResponse.newBuilder().setSyncContext(syncContext).build());
+
+        // should enqueue data payload
+        BlockingQueue<QueuePayload> streamQueue = connector.getStreamQueue();
+        QueuePayload payload = streamQueue.poll(1000, TimeUnit.MILLISECONDS);
+        assertNotNull(payload);
+        assertEquals(syncContext, payload.getSyncContext());
+        assertEquals(QueuePayloadType.DATA, payload.getType());
+        // should NOT have restarted the stream (1 call)
+        verify(stub, times(1)).syncFlags(any(), any());
+    }
+
     @Test
     void onErrorEnqueuesDataPayload() throws Exception {
         SyncStreamQueueSource connector =