add log and cleanup failed non-durable subscription (#444)

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

@@ -1595,9 +1595,7 @@ public class ManagedCursorImpl implements ManagedCursor {
             PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
             PositionImpl oldReadPosition = readPosition;
 
-            if (log.isDebugEnabled()) {
-                log.debug("Rewind from {} to {}", oldReadPosition, newReadPosition);
-            }
+            log.info("[{}] Rewind from {} to {}", name, oldReadPosition, newReadPosition);
 
             readPosition = newReadPosition;
         } finally {

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/persistent/PersistentTopic.java

@@ -446,6 +446,9 @@ public class PersistentTopic implements Topic, AddEntryCallback {
 
         if (!subscriptionFuture.isDone()) {
             subscriptionFuture.complete(subscription);
+        } else {
+            // failed to initialize managed-cursor: clean up created subscription
+            subscriptions.remove(subscriptionName);
         }
 
         return subscriptionFuture;