Merge pull request #975 from zalando/ARUHA-1954

Added updated_at to subscription

database/migration/aruha-1954-subscription-add-field.sql

@@ -0,0 +1,4 @@
+SET ROLE zalando_nakadi_data_owner;
+
+UPDATE zn_data.subscription SET s_subscription_object = jsonb_set(s_subscription_object, '{updated_at}',
+                                      concat('"', s_subscription_object ->>'created_at', '"')::jsonb, true);

docs/_data/nakadi-event-bus-api.yaml

@@ -1042,7 +1042,8 @@ paths:
         - oauth2: ['nakadi.event_stream.read']
       description: |
         This endpoint only allows to update the authorization section of a subscription. All other properties are
-        immutable. This operation is restricted to subjects with administrative role.
+        immutable. This operation is restricted to subjects with administrative role. This call captures the timestamp
+        of the update request.
       parameters:
         - name: subscription
           in: body
@@ -2423,6 +2424,15 @@ definitions:
           specified when creating subscription and sending it may result in a client error.
         format: RFC 3339 date-time
         example: '1996-12-19T16:39:57-08:00'
+      updated_at:
+        type: string
+        readOnly: true
+        description: |
+          Timestamp of last update of the subscription. This is generated by Nakadi. It should not be
+          specified when creating subscription and sending it may result in a client error. Its initial value is same
+          as created_at.
+        format: RFC 3339 date-time
+        example: '1996-12-19T16:39:57-08:00'
       read_from:
         type: string
         description: |

src/acceptance-test/java/org/zalando/nakadi/webservice/hila/SubscriptionAT.java

@@ -19,7 +19,9 @@
 import org.zalando.nakadi.domain.ItemsWrapper;
 import org.zalando.nakadi.domain.PaginationLinks;
 import org.zalando.nakadi.domain.PaginationWrapper;
+import org.zalando.nakadi.domain.ResourceAuthorizationAttribute;
 import org.zalando.nakadi.domain.Subscription;
+import org.zalando.nakadi.domain.SubscriptionAuthorization;
 import org.zalando.nakadi.domain.SubscriptionBase;
 import org.zalando.nakadi.domain.SubscriptionEventTypeStats;
 import org.zalando.nakadi.utils.JsonTestHelper;
@@ -35,6 +37,7 @@
 import org.zalando.problem.Problem;
 
 import java.io.IOException;
+import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.stream.Collectors;
@@ -114,7 +117,8 @@ public void testSubscriptionBaseOperations() throws IOException {
 
         // retrieve subscription object from response
         final Subscription subFirst = MAPPER.readValue(response.print(), Subscription.class);
-
+        //check initialization of updated_At
+        assertThat(subFirst.getUpdatedAt(), equalTo(subFirst.getCreatedAt()));
         // when we try to create that subscription again - we should get status 200
         // and the subscription that already exists should be returned
         response = given()
@@ -137,11 +141,32 @@ public void testSubscriptionBaseOperations() throws IOException {
         response.then().statusCode(HttpStatus.SC_OK).contentType(JSON);
         final Subscription gotSubscription = MAPPER.readValue(response.print(), Subscription.class);
         assertThat(gotSubscription, equalTo(subFirst));
+
+        //Check for update time of the subscription
+        final Subscription updateSub = subFirst;
+        updateSub.setAuthorization(new SubscriptionAuthorization(
+                Collections.singletonList(new ResourceAuthorizationAttribute("user", "me")),
+                Collections.singletonList(new ResourceAuthorizationAttribute("user", "me"))));
+        final String updatedSubscription = MAPPER.writeValueAsString(updateSub);
+
+        response = given()
+                .body(updatedSubscription)
+                .contentType(JSON)
+                .put(format(SUBSCRIPTION_URL, subFirst.getId()));
+
+        response
+                .then()
+                .statusCode(HttpStatus.SC_NO_CONTENT);
+
+        response = get(format(SUBSCRIPTION_URL, subFirst.getId()));
+        response.then().statusCode(HttpStatus.SC_OK).contentType(JSON);
+        final Subscription updatedSub = MAPPER.readValue(response.print(), Subscription.class);
+        assertThat(updatedSub.getUpdatedAt(), not(equalTo(subFirst.getUpdatedAt())));
     }
 
     @Test
     public void testSubscriptionWithNullAuthorisation() {
-        final  EventType eventType = createEventType();
+        final EventType eventType = createEventType();
         final String subscription = "{\"owning_application\":\"app\",\"event_types\":[\""
                 + eventType.getName() + "\"], \"read_from\": \"end\", \"consumer_group\":\"test\"," +
                 "\"authorization\": {\"admins\": [], \"readers\": []}}";

src/main/java/org/zalando/nakadi/domain/Subscription.java

@@ -13,19 +13,31 @@ public Subscription() {
         super();
     }
 
-    public Subscription(final String id, final DateTime createdAt, final SubscriptionBase subscriptionBase) {
+    public Subscription(final String id, final DateTime createdAt, final DateTime updatedAt,
+                        final SubscriptionBase subscriptionBase) {
         super(subscriptionBase);
         this.id = id;
         this.createdAt = createdAt;
+        this.updatedAt = updatedAt;
     }
 
     private String id;
 
     private DateTime createdAt;
 
+    private DateTime updatedAt;
+
     @JsonInclude(JsonInclude.Include.NON_NULL)
     private List<SubscriptionEventTypeStats> status;
 
+    public DateTime getUpdatedAt() {
+        return updatedAt;
+    }
+
+    public void setUpdatedAt(final DateTime updatedAt) {
+        this.updatedAt = updatedAt;
+    }
+
     public String getId() {
         return id;
     }

src/main/java/org/zalando/nakadi/repository/db/SubscriptionDbRepository.java

@@ -62,7 +62,7 @@ public Subscription createSubscription(final SubscriptionBase subscriptionBase)
             final String newId = uuidGenerator.randomUUID().toString();
             final String keyFieldsHash = hashGenerator.generateSubscriptionKeyFieldsHash(subscriptionBase);
             final DateTime createdAt = new DateTime(DateTimeZone.UTC);
-            final Subscription subscription = new Subscription(newId, createdAt, subscriptionBase);
+            final Subscription subscription = new Subscription(newId, createdAt, createdAt, subscriptionBase);
 
             jdbcTemplate.update("INSERT INTO zn_data.subscription (s_id, s_subscription_object, s_key_fields_hash) " +
                             "VALUES (?, ?::JSONB, ?)",

src/main/java/org/zalando/nakadi/service/subscription/SubscriptionService.java

@@ -2,6 +2,8 @@
 
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
 import org.json.JSONObject;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -54,6 +56,7 @@
 import org.zalando.nakadi.util.SubscriptionsUriHelper;
 import org.zalando.nakadi.view.SubscriptionCursorWithoutToken;
 
+
 import javax.annotation.Nullable;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -144,6 +147,7 @@ public Subscription updateSubscription(final String subscriptionId, final Subscr
 
         subscriptionValidationService.validateSubscriptionChange(old, newValue);
         old.mergeFrom(newValue);
+        old.setUpdatedAt(new DateTime(DateTimeZone.UTC));
         subscriptionRepository.updateSubscription(old);
         return old;
     }

src/test/java/org/zalando/nakadi/controller/PostSubscriptionControllerTest.java

@@ -86,8 +86,8 @@ public void whenSubscriptionCreationIsDisabledThenCreationFails() throws Excepti
     @Test
     public void whenSubscriptionCreationDisabledThenReturnExistentSubscription() throws Exception {
         final SubscriptionBase subscriptionBase = builder().buildSubscriptionBase();
-        final Subscription existingSubscription = new Subscription("123", new DateTime(DateTimeZone.UTC),
-                subscriptionBase);
+        final DateTime createdAt = new DateTime(DateTimeZone.UTC);
+        final Subscription existingSubscription = new Subscription("123", createdAt, createdAt, subscriptionBase);
         existingSubscription.setReadFrom(SubscriptionBase.InitialPosition.BEGIN);
 
         when(subscriptionService.getExistingSubscription(any())).thenReturn(existingSubscription);
@@ -104,7 +104,8 @@ public void whenSubscriptionCreationDisabledThenReturnExistentSubscription() thr
     @Test
     public void whenPostValidSubscriptionThenOk() throws Exception {
         final SubscriptionBase subscriptionBase = builder().buildSubscriptionBase();
-        final Subscription subscription = new Subscription("123", new DateTime(DateTimeZone.UTC), subscriptionBase);
+        final DateTime createdAt = new DateTime(DateTimeZone.UTC);
+        final Subscription subscription = new Subscription("123", createdAt, createdAt, subscriptionBase);
 
         when(subscriptionService.getExistingSubscription(any())).thenThrow(new NoSuchSubscriptionException("", null));
         when(subscriptionService.createSubscription(any())).thenReturn(subscription);
@@ -191,8 +192,8 @@ public void whenEventTypeDoesNotExistThenUnprocessableEntity() throws Exception
     @Test
     public void whenSubscriptionExistsThenReturnIt() throws Exception {
         final SubscriptionBase subscriptionBase = builder().buildSubscriptionBase();
-        final Subscription existingSubscription = new Subscription("123", new DateTime(DateTimeZone.UTC),
-                subscriptionBase);
+        final DateTime createdAt = new DateTime(DateTimeZone.UTC);
+        final Subscription existingSubscription = new Subscription("123", createdAt, createdAt, subscriptionBase);
 
         when(subscriptionService.getExistingSubscription(any())).thenReturn(existingSubscription);
         when(subscriptionService.createSubscription(any())).thenThrow(new NoSuchEventTypeException("msg"));

src/test/java/org/zalando/nakadi/controller/SubscriptionControllerTest.java

@@ -140,6 +140,7 @@ public SubscriptionControllerTest() throws Exception {
     @Test
     public void whenGetSubscriptionThenOk() throws Exception {
         final Subscription subscription = builder().build();
+        subscription.setUpdatedAt(subscription.getCreatedAt());
         when(subscriptionRepository.getSubscription(subscription.getId())).thenReturn(subscription);
 
         getSubscription(subscription.getId())
@@ -150,6 +151,7 @@ public void whenGetSubscriptionThenOk() throws Exception {
     @Test
     public void whenGetNoneExistingSubscriptionThenNotFound() throws Exception {
         final Subscription subscription = builder().build();
+        subscription.setUpdatedAt(subscription.getCreatedAt());
         when(subscriptionRepository.getSubscription(subscription.getId()))
                 .thenThrow(new NoSuchSubscriptionException("dummy-message"));
         final ThrowableProblem expectedProblem = Problem.valueOf(NOT_FOUND, "dummy-message");
@@ -237,6 +239,7 @@ public void whenGetSubscriptionAndExceptionThenServiceUnavailable() throws Excep
     @Test
     public void whenGetSubscriptionStatThenOk() throws Exception {
         final Subscription subscription = builder().withEventType(TIMELINE.getEventType()).build();
+        subscription.setUpdatedAt(subscription.getCreatedAt());
         final Collection<Partition> partitions = Collections.singleton(
                 new Partition(TIMELINE.getEventType(), "0", "xz", null, Partition.State.ASSIGNED));
         final ZkSubscriptionNode zkSubscriptionNode =
@@ -278,6 +281,7 @@ public void whenGetSubscriptionStatThenOk() throws Exception {
     @SuppressWarnings("unchecked")
     public void whenGetSubscriptionNoEventTypesThenStatEmpty() throws Exception {
         final Subscription subscription = builder().withEventType("myET").build();
+        subscription.setUpdatedAt(subscription.getCreatedAt());
         when(subscriptionRepository.getSubscription(subscription.getId())).thenReturn(subscription);
         when(zkSubscriptionClient.getZkSubscriptionNode()).thenReturn(
                 Optional.of(new ZkSubscriptionNode(Collections.emptyList(), Collections.emptyList())));
@@ -326,6 +330,7 @@ private void mockGetFromRepoSubscriptionWithOwningApp(final String subscriptionI
                 .withId(subscriptionId)
                 .withOwningApplication(owningApplication)
                 .build();
+        subscription.setUpdatedAt(subscription.getCreatedAt());
         when(subscriptionRepository.getSubscription(subscriptionId)).thenReturn(subscription);
     }
 

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

@@ -65,6 +65,7 @@ public void whenSubscriptionCreatedThenKPIEventSubmitted() {
         final Subscription subscription = RandomSubscriptionBuilder.builder()
                 .withId("my_subscription_id1")
                 .build();
+        subscription.setUpdatedAt(subscription.getCreatedAt());
         when(subscriptionRepository.createSubscription(subscriptionBase)).thenReturn(subscription);
 
         subscriptionService.createSubscription(subscriptionBase);