Merge branch 'master' into ARUHA-1997

docs/_documentation/faq.md

@@ -83,8 +83,6 @@ The default behavior when running the docker containers locally will be for OAut
 
 If you are running a Nakadi server locally outside docker, you can disable token checks by setting the environment variable `NAKADI_OAUTH2_MODE` to `OFF` before starting the server.
 
-Note that, even if OAuth is disabled using the `NAKADI_OAUTH2_MODE` environment variable, the current behavior will be to check a token if one is sent by a client so you might need to configure the client to also not send tokens.
-
 #### I want to send arbitrary JSON, how do I avoid defining a JSON Schema?
 
 The standard workaround is to define an event type with the following category and schema:

src/main/java/org/zalando/nakadi/config/SecurityConfiguration.java

@@ -10,6 +10,9 @@
 import org.springframework.http.converter.HttpMessageNotWritableException;
 import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter;
 import org.springframework.security.config.annotation.web.builders.HttpSecurity;
+import org.springframework.security.config.annotation.web.builders.WebSecurity;
+import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity;
+import org.springframework.security.config.annotation.web.configuration.WebSecurityConfigurerAdapter;
 import org.springframework.security.core.AuthenticationException;
 import org.springframework.security.oauth2.common.exceptions.OAuth2Exception;
 import org.springframework.security.oauth2.config.annotation.web.configuration.EnableResourceServer;
@@ -201,4 +204,16 @@ private static Status fromStatusCode(final int code) throws UnknownStatusCodeExc
         }
         throw new UnknownStatusCodeException("Unknown status code: " + code);
     }
+
+    @Configuration
+    @EnableWebSecurity
+    public class WebSecurityConfiguration extends WebSecurityConfigurerAdapter {
+        @Override
+        public void configure(final WebSecurity web) throws Exception {
+            if (settings.getAuthMode() == SecuritySettings.AuthMode.OFF) {
+                web.ignoring().anyRequest();
+            }
+        }
+    }
+
 }

src/main/java/org/zalando/nakadi/service/subscription/state/PartitionData.java

@@ -112,6 +112,14 @@ private List<ConsumedEvent> extract(final int count) {
         return result;
     }
 
+    public List<ConsumedEvent> extractMaxEvents(final long currentTimeMillis, final int count) {
+        final List<ConsumedEvent> result = extract(count);
+        if(!result.isEmpty()) {
+            lastSendMillis = currentTimeMillis;
+        }
+        return result;
+    }
+
     int getKeepAliveInARow() {
         return keepAliveInARow;
     }

src/main/java/org/zalando/nakadi/service/subscription/state/StreamingState.java

@@ -252,15 +252,16 @@ private void streamToOutput(final boolean streamTimeoutReached) {
         }
 
         long memoryConsumed = offsets.values().stream().mapToLong(PartitionData::getBytesInMemory).sum();
-        while (memoryConsumed > getContext().getStreamMemoryLimitBytes()) {
+        while (memoryConsumed > getContext().getStreamMemoryLimitBytes() && getMessagesAllowedToSend() > 0) {
             // Select heaviest guy (and on previous step we figured out that we can not send anymore full batches,
             // therefore we can take all the events from one partition.
             final Map.Entry<EventTypePartition, PartitionData> heaviestPartition = offsets.entrySet().stream().max(
                     Comparator.comparing(e -> e.getValue().getBytesInMemory())
             ).get(); // There is always at least 1 item in list
 
             long deltaSize = heaviestPartition.getValue().getBytesInMemory();
-            final List<ConsumedEvent> events = heaviestPartition.getValue().extractAll(currentTimeMillis);
+            final List<ConsumedEvent> events = heaviestPartition.getValue().extractMaxEvents(currentTimeMillis,
+                    (int) getMessagesAllowedToSend());
             deltaSize -= heaviestPartition.getValue().getBytesInMemory();
 
             sentSomething = true;

src/main/java/org/zalando/nakadi/service/subscription/zk/AbstractZkSubscriptionClient.java

@@ -14,11 +14,11 @@
 import org.zalando.nakadi.domain.EventTypePartition;
 import org.zalando.nakadi.exceptions.runtime.NakadiBaseException;
 import org.zalando.nakadi.exceptions.runtime.NakadiRuntimeException;
-import org.zalando.nakadi.exceptions.runtime.UnableProcessException;
 import org.zalando.nakadi.exceptions.runtime.OperationInterruptedException;
 import org.zalando.nakadi.exceptions.runtime.OperationTimeoutException;
 import org.zalando.nakadi.exceptions.runtime.RequestInProgressException;
 import org.zalando.nakadi.exceptions.runtime.ServiceTemporarilyUnavailableException;
+import org.zalando.nakadi.exceptions.runtime.UnableProcessException;
 import org.zalando.nakadi.exceptions.runtime.ZookeeperException;
 import org.zalando.nakadi.service.subscription.model.Session;
 import org.zalando.nakadi.view.SubscriptionCursorWithoutToken;
@@ -46,17 +46,17 @@
 import static org.echocat.jomon.runtime.concurrent.Retryer.executeWithRetry;
 
 public abstract class AbstractZkSubscriptionClient implements ZkSubscriptionClient {
+    public static final int SECONDS_TO_WAIT_FOR_LOCK = 15;
+    protected static final String NODE_TOPOLOGY = "/topology";
     private static final String STATE_INITIALIZED = "INITIALIZED";
     private static final int COMMIT_CONFLICT_RETRY_TIMES = 5;
-    protected static final String NODE_TOPOLOGY = "/topology";
-    public static final int SECONDS_TO_WAIT_FOR_LOCK = 15;
     private static final int MAX_ZK_RESPONSE_SECONDS = 5;
 
     private final String subscriptionId;
     private final CuratorFramework curatorFramework;
-    private InterProcessSemaphoreMutex lock;
     private final String resetCursorPath;
     private final Logger log;
+    private InterProcessSemaphoreMutex lock;
 
     public AbstractZkSubscriptionClient(
             final String subscriptionId,
@@ -206,6 +206,8 @@ protected <K, V> Map<K, V> loadDataAsync(final Collection<K> keys,
                             synchronized (result) {
                                 result.put(key, value);
                             }
+                        } else if (event.getResultCode() == KeeperException.Code.NONODE.intValue()) {
+                            getLog().warn("Unable to get {} data from zk. Node not found ", zkKey);
                         } else {
                             getLog().error(
                                     "Failed to get {} data from zk. status code: {}",
@@ -229,35 +231,29 @@ protected <K, V> Map<K, V> loadDataAsync(final Collection<K> keys,
             Thread.currentThread().interrupt();
             throw new ServiceTemporarilyUnavailableException("Failed to wait for zk response", ex);
         }
-        if (result.size() != keys.size()) {
-            throw new ServiceTemporarilyUnavailableException("Failed to wait for keys " +
-                    keys.stream()
-                            .filter(v -> !result.containsKey(v))
-                            .map(String::valueOf)
-                            .collect(Collectors.joining(", "))
-                    + " to be in response", null);
-        }
         return result;
     }
 
     @Override
     public final Collection<Session> listSessions()
             throws SubscriptionNotInitializedException, NakadiRuntimeException, ServiceTemporarilyUnavailableException {
         getLog().info("fetching sessions information");
-        final List<String> zkSessions;
-        try {
-            zkSessions = getCurator().getChildren().forPath(getSubscriptionPath("/sessions"));
-        } catch (final KeeperException.NoNodeException e) {
-            throw new SubscriptionNotInitializedException(getSubscriptionId());
-        } catch (Exception ex) {
-            throw new NakadiRuntimeException(ex);
+        for (int i = 0; i < 5; i++) {
+            try {
+                final List<String> sessions = getCurator().getChildren().forPath(getSubscriptionPath("/sessions"));
+                final Map <String,Session> result = loadDataAsync(sessions,
+                        key -> getSubscriptionPath("/sessions/" + key),
+                        this::deserializeSession);
+                if (result.size() == sessions.size()) {
+                    return result.values();
+                }
+            } catch (final KeeperException.NoNodeException e) {
+                throw new SubscriptionNotInitializedException(getSubscriptionId());
+            } catch (Exception ex) {
+                throw new NakadiRuntimeException(ex);
+            }
         }
-
-        return loadDataAsync(
-                zkSessions,
-                key -> getSubscriptionPath("/sessions/" + key),
-                this::deserializeSession
-        ).values();
+        throw new ServiceTemporarilyUnavailableException("Failed to get all keys from ZK", null);
     }
 
     @Override

src/main/java/org/zalando/nakadi/service/subscription/zk/NewZkSubscriptionClient.java

@@ -17,6 +17,7 @@
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import static com.google.common.base.Charsets.UTF_8;
@@ -170,8 +171,21 @@ protected String getOffsetPath(final EventTypePartition etp) {
     public Map<EventTypePartition, SubscriptionCursorWithoutToken> getOffsets(
             final Collection<EventTypePartition> keys)
             throws NakadiRuntimeException, ServiceTemporarilyUnavailableException {
-        return loadDataAsync(keys, this::getOffsetPath, (etp, value) ->
-                new SubscriptionCursorWithoutToken(etp.getEventType(), etp.getPartition(), new String(value, UTF_8)));
+        final Map<EventTypePartition, SubscriptionCursorWithoutToken> offSets = loadDataAsync(keys,
+                this::getOffsetPath, (etp, value) ->
+                        new SubscriptionCursorWithoutToken(etp.getEventType(), etp.getPartition(),
+                                new String(value, UTF_8)));
+
+        if (offSets.size() != keys.size()) {
+            throw new ServiceTemporarilyUnavailableException("Failed to get all the keys " +
+                    keys.stream()
+                            .filter(v -> !offSets.containsKey(v))
+                            .map(String::valueOf)
+                            .collect(Collectors.joining(", "))
+                    + " from ZK.", null);
+        }
+
+        return offSets;
     }
 
     @Override