Merge pull request #913 from zalando/log-compaction-feature-flag

Add Log compaction feature toggle

CHANGELOG.md

@@ -8,6 +8,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [2.8.1]
 
+### Added
+- Log Compaction Feature Toggle
+
 ### Changed
 - Upgraded Kafka client to 1.1.0
 

docker-compose.yml

@@ -23,6 +23,7 @@ services:
       - NAKADI_FEATURES_DEFAULT_FEATURES_REMOTE_TOKENINFO
       - NAKADI_FEATURES_DEFAULT_FEATURES_KPI_COLLECTION
       - NAKADI_FEATURES_DEFAULT_FEATURES_DISABLE_DB_WRITE_OPERATIONS
+      - NAKADI_FEATURES_DEFAULT_FEATURES_DISABLE_LOG_COMPACTION
       - NAKADI_ZOOKEEPER_BROKERS=localhost:2181
       - SPRING_DATASOURCE_URL=jdbc:postgresql://localhost:5432/local_nakadi_db
       - SPRING_DATASOURCE_USERNAME=nakadi

src/main/java/org/zalando/nakadi/controller/ExceptionHandling.java

@@ -10,6 +10,7 @@
 import org.springframework.web.bind.annotation.ControllerAdvice;
 import org.springframework.web.bind.annotation.ExceptionHandler;
 import org.springframework.web.context.request.NativeWebRequest;
+import org.zalando.nakadi.exceptions.runtime.FeatureNotAvailableException;
 import org.zalando.nakadi.exceptions.runtime.IllegalClientIdException;
 import org.zalando.nakadi.exceptions.NakadiException;
 import org.zalando.nakadi.exceptions.NakadiRuntimeException;
@@ -31,6 +32,7 @@
 
 import javax.ws.rs.core.Response;
 
+import static javax.ws.rs.core.Response.Status.NOT_IMPLEMENTED;
 import static org.zalando.problem.MoreStatus.UNPROCESSABLE_ENTITY;
 
 
@@ -173,4 +175,13 @@ public ResponseEntity<Problem> handleDbWriteOperationsBlockedException(
                 "Database is currently in read-only mode", request);
     }
 
+    @ExceptionHandler(FeatureNotAvailableException.class)
+    public ResponseEntity<Problem> handleFeatureNotAvailable(
+            final FeatureNotAvailableException ex,
+            final NativeWebRequest request) {
+        LOG.debug(ex.getMessage(), ex);
+        return Responses.create(Problem.valueOf(NOT_IMPLEMENTED, ex.getMessage()), request);
+
+    }
+
 }

src/main/java/org/zalando/nakadi/controller/PostSubscriptionController.java

@@ -16,7 +16,6 @@
 import org.zalando.nakadi.domain.Subscription;
 import org.zalando.nakadi.domain.SubscriptionBase;
 import org.zalando.nakadi.exceptions.runtime.DuplicatedSubscriptionException;
-import org.zalando.nakadi.exceptions.runtime.FeatureNotAvailableException;
 import org.zalando.nakadi.exceptions.runtime.InconsistentStateException;
 import org.zalando.nakadi.exceptions.runtime.MyNakadiRuntimeException1;
 import org.zalando.nakadi.exceptions.runtime.NoEventTypeException;
@@ -26,8 +25,8 @@
 import org.zalando.nakadi.plugin.api.ApplicationService;
 import org.zalando.nakadi.problem.ValidationProblem;
 import org.zalando.nakadi.security.Client;
-import org.zalando.nakadi.service.subscription.SubscriptionService;
 import org.zalando.nakadi.service.FeatureToggleService;
+import org.zalando.nakadi.service.subscription.SubscriptionService;
 import org.zalando.problem.MoreStatus;
 import org.zalando.problem.Problem;
 import org.zalando.problem.spring.web.advice.Responses;
@@ -36,7 +35,6 @@
 import javax.ws.rs.core.HttpHeaders;
 import javax.ws.rs.core.Response;
 
-import static javax.ws.rs.core.Response.Status.NOT_IMPLEMENTED;
 import static org.springframework.http.HttpStatus.OK;
 import static org.zalando.nakadi.service.FeatureToggleService.Feature.CHECK_OWNING_APPLICATION;
 import static org.zalando.nakadi.service.FeatureToggleService.Feature.DISABLE_SUBSCRIPTION_CREATION;
@@ -113,13 +111,4 @@ public ResponseEntity<Problem> handleUnprocessableSubscription(final MyNakadiRun
         LOG.debug("Error occurred when working with subscriptions", exception);
         return Responses.create(MoreStatus.UNPROCESSABLE_ENTITY, exception.getMessage(), request);
     }
-
-    @ExceptionHandler(FeatureNotAvailableException.class)
-    public ResponseEntity<Problem> handleFeatureNotAvailable(
-            final FeatureNotAvailableException ex,
-            final NativeWebRequest request) {
-        LOG.debug(ex.getMessage(), ex);
-        return Responses.create(Problem.valueOf(NOT_IMPLEMENTED, ex.getMessage()), request);
-
-    }
 }

src/main/java/org/zalando/nakadi/service/EventTypeService.java

@@ -38,6 +38,7 @@
 import org.zalando.nakadi.exceptions.runtime.EventTypeDeletionException;
 import org.zalando.nakadi.exceptions.runtime.EventTypeOptionsValidationException;
 import org.zalando.nakadi.exceptions.runtime.EventTypeUnavailableException;
+import org.zalando.nakadi.exceptions.runtime.FeatureNotAvailableException;
 import org.zalando.nakadi.exceptions.runtime.InconsistentStateException;
 import org.zalando.nakadi.exceptions.runtime.InvalidEventTypeException;
 import org.zalando.nakadi.exceptions.runtime.NoEventTypeException;
@@ -146,6 +147,11 @@ public void create(final EventTypeBase eventType)
             throw new DbWriteOperationsBlockedException("Cannot create event type: write operations on DB " +
                     "are blocked by feature flag.");
         }
+        if (eventType.getCleanupPolicy() == CleanupPolicy.COMPACT
+                && featureToggleService.isFeatureEnabled(FeatureToggleService.Feature.DISABLE_LOG_COMPACTION)) {
+            throw new FeatureNotAvailableException("log compaction is not available",
+                    FeatureToggleService.Feature.DISABLE_LOG_COMPACTION);
+        }
         eventTypeOptionsValidator.checkRetentionTime(eventType.getOptions());
         setDefaultEventTypeOptions(eventType);
         validateSchema(eventType);

src/main/java/org/zalando/nakadi/service/FeatureToggleService.java

@@ -39,7 +39,8 @@ enum Feature {
         SEND_BATCH_VIA_OUTPUT_STREAM("send_batch_via_output_stream"),
         REMOTE_TOKENINFO("remote_tokeninfo"),
         KPI_COLLECTION("kpi_collection"),
-        DISABLE_DB_WRITE_OPERATIONS("disable_db_write_operations");
+        DISABLE_DB_WRITE_OPERATIONS("disable_db_write_operations"),
+        DISABLE_LOG_COMPACTION("disable_log_compaction");
 
         private final String id;
 

src/test/java/org/zalando/nakadi/service/EventTypeServiceTest.java

@@ -9,12 +9,14 @@
 import org.springframework.transaction.support.TransactionCallback;
 import org.springframework.transaction.support.TransactionTemplate;
 import org.zalando.nakadi.config.NakadiSettings;
+import org.zalando.nakadi.domain.CleanupPolicy;
 import org.zalando.nakadi.domain.EventType;
 import org.zalando.nakadi.domain.Subscription;
 import org.zalando.nakadi.enrichment.Enrichment;
 import org.zalando.nakadi.exceptions.InternalNakadiException;
 import org.zalando.nakadi.exceptions.runtime.ConflictException;
 import org.zalando.nakadi.exceptions.runtime.EventTypeDeletionException;
+import org.zalando.nakadi.exceptions.runtime.FeatureNotAvailableException;
 import org.zalando.nakadi.exceptions.runtime.TopicCreationException;
 import org.zalando.nakadi.partitioning.PartitionResolver;
 import org.zalando.nakadi.repository.EventTypeRepository;
@@ -130,6 +132,17 @@ public void testFeatureToggleAllowsDeleteEventTypeWithSubscriptions() throws Exc
         // no exception should be thrown
     }
 
+    @Test(expected = FeatureNotAvailableException.class)
+    public void testFeatureToggleDisableLogCompaction() throws Exception {
+        final EventType eventType = buildDefaultEventType();
+        eventType.setCleanupPolicy(CleanupPolicy.COMPACT);
+
+        when(featureToggleService.isFeatureEnabled(FeatureToggleService.Feature.DISABLE_LOG_COMPACTION))
+                .thenReturn(true);
+
+        eventTypeService.create(eventType);
+    }
+
     @Test
     public void shouldRemoveEventTypeWhenTimelineCreationFails() throws Exception {
         final EventType eventType = buildDefaultEventType();