Merge branch 'main' into add-flag-push

.github/workflows/lint.yml

@@ -19,5 +19,5 @@ jobs:
           distribution: 'zulu'
           java-version: '8'
 
-    - name: Lint
-      run: ./gradlew ktlintCheck
+      - name: Lint
+        run: ./gradlew ktlintCheck

CHANGELOG.md

@@ -1,3 +1,36 @@
+## [1.4.3](https://github.com/amplitude/experiment-jvm-server/compare/1.4.2...1.4.3) (2024-09-23)
+
+
+### Bug Fixes
+
+* add metric for cohort download too large ([#33](https://github.com/amplitude/experiment-jvm-server/issues/33)) ([5a855c9](https://github.com/amplitude/experiment-jvm-server/commit/5a855c904b011b931467a29f48914bd7867f83db))
+
+## [1.4.2](https://github.com/amplitude/experiment-jvm-server/compare/1.4.1...1.4.2) (2024-09-20)
+
+
+### Bug Fixes
+
+* fix assignment config to use batch when configured ([#32](https://github.com/amplitude/experiment-jvm-server/issues/32)) ([d6d9684](https://github.com/amplitude/experiment-jvm-server/commit/d6d9684332066aaf5345646061651e67a7ea1247))
+
+## [1.4.1](https://github.com/amplitude/experiment-jvm-server/compare/1.4.0...1.4.1) (2024-09-04)
+
+
+### Bug Fixes
+
+* fix certain metrics that weren't firing ([dbf2994](https://github.com/amplitude/experiment-jvm-server/commit/dbf299481ab5c2d2f985edb0219a427d5b6a35ee))
+
+# [1.4.0](https://github.com/amplitude/experiment-jvm-server/compare/1.3.1...1.4.0) (2024-08-27)
+
+
+### Bug Fixes
+
+* use get request for remote fetch ([#29](https://github.com/amplitude/experiment-jvm-server/issues/29)) ([6c2fc3a](https://github.com/amplitude/experiment-jvm-server/commit/6c2fc3ab7d152cdbfbcccf8d6dee761b9d3ffe73))
+
+
+### Features
+
+* local evaluation cohorts support ([#28](https://github.com/amplitude/experiment-jvm-server/issues/28)) ([d2d5213](https://github.com/amplitude/experiment-jvm-server/commit/d2d5213b57f60ed5d032ebcbbd91487ec8ca7fc9))
+
 ## [1.3.1](https://github.com/amplitude/experiment-jvm-server/compare/1.3.0...1.3.1) (2024-05-23)
 
 

build.gradle.kts

@@ -33,7 +33,7 @@ dependencies {
 // Publishing
 
 group = "com.amplitude"
-version = "1.3.1"
+version = "1.4.3"
 
 nexusPublishing {
     repositories {

src/main/kotlin/Experiment.kt

@@ -4,7 +4,7 @@ import com.amplitude.experiment.util.Logger
 import com.amplitude.experiment.util.SystemLogger
 import java.util.concurrent.Executors
 
-internal const val LIBRARY_VERSION = "1.3.1"
+internal const val LIBRARY_VERSION = "1.4.3"
 
 object Experiment {
 

src/main/kotlin/LocalEvaluationClient.kt

@@ -43,14 +43,14 @@ class LocalEvaluationClient internal constructor(
     apiKey: String,
     private val config: LocalEvaluationConfig = LocalEvaluationConfig(),
     private val httpClient: OkHttpClient = OkHttpClient(),
-    cohortApi: CohortApi? = getCohortDownloadApi(config, httpClient)
+    private val metrics: LocalEvaluationMetrics = LocalEvaluationMetricsWrapper(config.metrics),
+    cohortApi: CohortApi? = getCohortDownloadApi(config, httpClient, metrics),
 ) {
     private val assignmentService: AssignmentService? = createAssignmentService(apiKey)
     private val serverUrl: HttpUrl = getServerUrl(config)
     private val streamServerUrl: HttpUrl = getStreamServerUrl(config)
     private val evaluation: EvaluationEngine = EvaluationEngineImpl()
-    private val metrics: LocalEvaluationMetrics = LocalEvaluationMetricsWrapper(config.metrics)
-    private val flagConfigApi = DynamicFlagConfigApi(apiKey, serverUrl, null, httpClient)
+    private val flagConfigApi = DynamicFlagConfigApi(apiKey, serverUrl, null, httpClient, metrics)
     private val proxyUrl: HttpUrl? = getProxyUrl(config)
     private val flagConfigProxyApi = if (proxyUrl == null) null else DynamicFlagConfigApi(apiKey, proxyUrl, null, httpClient)
     private val flagConfigStreamApi = if (config.streamUpdates) FlagConfigStreamApi(apiKey, streamServerUrl, httpClient, config.streamFlagConnTimeoutMillis) else null
@@ -96,11 +96,11 @@ class LocalEvaluationClient internal constructor(
                 init(config.assignmentConfiguration.apiKey)
                 setEventUploadThreshold(config.assignmentConfiguration.eventUploadThreshold)
                 setEventUploadPeriodMillis(config.assignmentConfiguration.eventUploadPeriodMillis)
-                useBatchMode(config.assignmentConfiguration.useBatchMode)
                 setOptions(Options().setMinIdLength(1))
                 setServerUrl(getEventServerUrl(config, config.assignmentConfiguration))
             },
-            InMemoryAssignmentFilter(config.assignmentConfiguration.cacheCapacity)
+            InMemoryAssignmentFilter(config.assignmentConfiguration.cacheCapacity),
+            metrics = metrics,
         )
     }
     @JvmOverloads
@@ -171,7 +171,11 @@ class LocalEvaluationClient internal constructor(
     }
 }
 
-private fun getCohortDownloadApi(config: LocalEvaluationConfig, httpClient: OkHttpClient): CohortApi? {
+private fun getCohortDownloadApi(
+    config: LocalEvaluationConfig,
+    httpClient: OkHttpClient,
+    metrics: LocalEvaluationMetrics
+): CohortApi? {
     return if (config.cohortSyncConfig != null) {
         DynamicCohortApi(
             apiKey = config.cohortSyncConfig.apiKey,
@@ -180,6 +184,7 @@ private fun getCohortDownloadApi(config: LocalEvaluationConfig, httpClient: OkHt
             serverUrl = getCohortServerUrl(config),
             proxyUrl = getProxyUrl(config),
             httpClient = httpClient,
+            metrics = metrics
         )
     } else {
         null
@@ -230,8 +235,16 @@ private fun getEventServerUrl(
 ): String {
     return if (assignmentConfiguration.serverUrl == LocalEvaluationConfig.Defaults.EVENT_SERVER_URL) {
         when (config.serverZone) {
-            ServerZone.US -> US_EVENT_SERVER_URL
-            ServerZone.EU -> EU_EVENT_SERVER_URL
+            ServerZone.US -> if (assignmentConfiguration.useBatchMode) {
+                US_BATCH_SERVER_URL
+            } else {
+                US_EVENT_SERVER_URL
+            }
+            ServerZone.EU -> if (assignmentConfiguration.useBatchMode) {
+                EU_BATCH_SERVER_URL
+            } else {
+                EU_EVENT_SERVER_URL
+            }
         }
     } else {
         assignmentConfiguration.serverUrl

src/main/kotlin/LocalEvaluationConfig.kt

@@ -240,6 +240,7 @@ interface LocalEvaluationMetrics {
     fun onFlagConfigStream()
     fun onFlagConfigStreamFailure(exception: Exception?)
     fun onCohortDownload()
+    fun onCohortDownloadTooLarge(exception: Exception)
     fun onCohortDownloadFailure(exception: Exception)
     fun onCohortDownloadOriginFallback(exception: Exception)
     fun onProxyCohortMembership()

src/main/kotlin/RemoteEvaluationClient.kt

@@ -13,10 +13,8 @@ import okhttp3.Call
 import okhttp3.Callback
 import okhttp3.HttpUrl
 import okhttp3.HttpUrl.Companion.toHttpUrl
-import okhttp3.MediaType.Companion.toMediaType
 import okhttp3.OkHttpClient
 import okhttp3.Request
-import okhttp3.RequestBody.Companion.toRequestBody
 import okhttp3.Response
 import okio.IOException
 import java.util.Base64

src/main/kotlin/ServerZone.kt

@@ -8,6 +8,8 @@ internal const val US_COHORT_SERVER_URL = "https://cohort-v2.lab.amplitude.com"
 internal const val EU_COHORT_SERVER_URL = "https://cohort-v2.lab.eu.amplitude.com"
 internal const val US_EVENT_SERVER_URL = "https://api2.amplitude.com/2/httpapi"
 internal const val EU_EVENT_SERVER_URL = "https://api.eu.amplitude.com/2/httpapi"
+internal const val US_BATCH_SERVER_URL = "https://api2.amplitude.com/batch"
+internal const val EU_BATCH_SERVER_URL = "https://api.eu.amplitude.com/batch"
 
 enum class ServerZone {
     US, EU

src/main/kotlin/cohort/CohortLoader.kt

@@ -44,7 +44,8 @@ internal class CohortLoader(
                     } catch (e: CohortNotModifiedException) {
                         // Do nothing
                     } catch (e: CohortTooLargeException) {
-                        Logger.e("Cohort too large", e)
+                        metrics.onCohortDownloadTooLarge(e)
+                        throw e
                     }
                 }
             }, executor).whenComplete { _, _ -> jobs.remove(cohortId) }

src/main/kotlin/deployment/DeploymentRunner.kt

@@ -35,7 +35,7 @@ internal class DeploymentRunner(
     private val lock = Once()
     private val poller = Executors.newScheduledThreadPool(1, daemonFactory)
     private val cohortLoader = if (cohortApi != null && cohortStorage != null) {
-        CohortLoader(cohortApi, cohortStorage)
+        CohortLoader(cohortApi, cohortStorage, metrics)
     } else {
         null
     }
@@ -71,12 +71,17 @@ internal class DeploymentRunner(
                     try {
                         val cohortIds = flagConfigStorage.getFlagConfigs().values.getAllCohortIds()
                         for (cohortId in cohortIds) {
-                            cohortLoader.loadCohort(cohortId)
+                            cohortLoader.loadCohort(cohortId).handle { _, exception ->
+                                if (exception != null) {
+                                    Logger.e("Failed to load cohort $cohortId", exception)
+                                }
+                            }
                         }
                     } catch (t: Throwable) {
                         Logger.e("Refresh cohorts failed.", t)
                     }
-                }, cohortPollingInterval,
+                },
+                cohortPollingInterval,
                 cohortPollingInterval,
                 TimeUnit.MILLISECONDS
             )

src/main/kotlin/util/Metrics.kt

@@ -86,6 +86,11 @@ internal class LocalEvaluationMetricsWrapper(
         executor?.execute { metrics.onCohortDownload() }
     }
 
+    override fun onCohortDownloadTooLarge(exception: Exception) {
+        val metrics = metrics ?: return
+        executor?.execute { metrics.onCohortDownloadTooLarge(exception) }
+    }
+
     override fun onCohortDownloadFailure(exception: Exception) {
         val metrics = metrics ?: return
         executor?.execute { metrics.onCohortDownloadFailure(exception) }

src/test/kotlin/cohort/CohortLoaderTest.kt

@@ -35,7 +35,11 @@ class CohortLoaderTest {
         `when`(api.getCohort("b", null)).thenReturn(cohortB)
         val storage = InMemoryCohortStorage()
         val loader = CohortLoader(api, storage)
-        loader.loadCohort("a").get()
+        try {
+            loader.loadCohort("a").get()
+        } catch (t: Throwable) {
+            // Expected
+        }
         loader.loadCohort("b").get()
         val storageDescriptionA = storage.getCohort("a")
         val storageDescriptionB = storage.getCohort("b")