Merge pull request #625 from CJSCommonPlatform/release-4.1.x-fix-for-stream-status-update

fix for source update

Author: purple52

event-buffer/event-buffer-core/src/main/java/uk/gov/justice/services/event/buffer/core/repository/streamstatus/StreamStatusJdbcRepository.java

@@ -31,11 +31,12 @@ public class StreamStatusJdbcRepository {
     /**
      * Statements
      */
-    private static final String SELECT_BY_STREAM_ID_AND_SOURCE = "SELECT stream_id, version, source FROM stream_status WHERE stream_id=? AND source=? FOR UPDATE";
+    private static final String SELECT_BY_STREAM_ID_AND_SOURCE = "SELECT stream_id, version, source FROM stream_status WHERE stream_id=? AND source in (?,'unknown') FOR UPDATE";
+    private static final String COUNT_BY_STREAM_ID = "SELECT count(*) FROM stream_status WHERE stream_id=?";
     private static final String INSERT = "INSERT INTO stream_status (version, stream_id, source) VALUES (?, ?, ?)";
     private static final String INSERT_ON_CONFLICT_DO_NOTHING = new StringBuilder().append(INSERT).append(" ON CONFLICT DO NOTHING").toString();
     private static final String UPDATE = "UPDATE stream_status SET version=?,source=? WHERE stream_id=? and source in (?,'unknown')";
-
+    private static final String UPDATE_UNKNOWN_SOURCE = "UPDATE stream_status SET source=? WHERE stream_id=? and source = 'unknown'";
 
     @Inject
     JdbcRepositoryHelper jdbcRepositoryHelper;
@@ -126,6 +127,25 @@ public Optional<StreamStatus> findByStreamIdAndSource(final UUID streamId, final
         }
     }
 
+    /**
+     * Returns a count of records for a given stream streamId.
+     *
+     * @param streamId streamId of the stream.
+     * @return a int.
+     */
+    public int countByStreamId(final UUID streamId) {
+        try (final PreparedStatementWrapper ps = jdbcRepositoryHelper.preparedStatementWrapperOf(dataSource, COUNT_BY_STREAM_ID)) {
+            ps.setObject(1, streamId);
+            final ResultSet resultSet = ps.executeQuery();
+            while (resultSet.next()) {
+                return resultSet.getInt(1);
+            }
+        } catch (SQLException e) {
+            throw new JdbcRepositoryException(format("Exception while looking up status of the stream: %s", streamId), e);
+        }
+        return 0;
+    }
+
     private Optional<StreamStatus> streamStatusFrom(final PreparedStatementWrapper ps) throws SQLException {
         final ResultSet resultSet = ps.executeQuery();
         return resultSet.next()
@@ -137,4 +157,14 @@ private Optional<StreamStatus> streamStatusFrom(final PreparedStatementWrapper p
     protected StreamStatus entityFrom(final ResultSet rs) throws SQLException {
         return new StreamStatus((UUID) rs.getObject(PRIMARY_KEY_ID), rs.getLong(COL_VERSION), rs.getString(SOURCE));
     }
+
+    public void updateSource(final UUID streamId, final String source) {
+        try (final PreparedStatementWrapper ps = jdbcRepositoryHelper.preparedStatementWrapperOf(dataSource, UPDATE_UNKNOWN_SOURCE)) {
+            ps.setString(1, source);
+            ps.setObject(2, streamId);
+            ps.executeUpdate();
+        } catch (SQLException e) {
+            throw new JdbcRepositoryException(format("Exception while updating unknown source of the stream: %s", streamId), e);
+        }
+    }
 }

event-buffer/event-buffer-core/src/main/java/uk/gov/justice/services/event/buffer/core/service/PostgreSQLBasedBufferInitialisationStrategy.java

@@ -16,6 +16,7 @@ public PostgreSQLBasedBufferInitialisationStrategy(final StreamStatusJdbcReposit
 
     @Override
     public long initialiseBuffer(final UUID streamId, final String source) {
+        streamStatusRepository.updateSource(streamId,source);
         streamStatusRepository.insertOrDoNothing(new StreamStatus(streamId, INITIAL_VERSION, source));
         return streamStatusRepository.findByStreamIdAndSource(streamId, source)
                 .orElseThrow(() -> new IllegalStateException("stream status cannot be empty"))

event-buffer/event-buffer-core/src/test/java/uk/gov/justice/services/event/buffer/core/repository/streamstatus/StreamStatusJdbcRepositoryIT.java

@@ -25,6 +25,7 @@ public class StreamStatusJdbcRepositoryIT {
 
     private static final String LIQUIBASE_STREAM_STATUS_CHANGELOG_XML = "liquibase/event-buffer-changelog.xml";
 
+    private static final long INITIAL_VERSION = 0L;
 
     private StreamStatusJdbcRepository jdbcRepository;
 
@@ -53,7 +54,25 @@ public void initDatabase() throws Exception {
         }
     }
 
+    @Test
+    public void shouldNotCreateNewStreamStatusRecordWhenANewSourceIsIntroduced() throws Exception {
+        final String source = "unknown";
+        final UUID streamId = randomUUID();
+
+        initialiseBuffer(streamId, "unknown");
+        final StreamStatus streamStatus = new StreamStatus(streamId, 2L, source);
+
+        jdbcRepository.update(streamStatus);
+
+        initialiseBuffer(streamId, "sjp");
+        final int count = jdbcRepository.countByStreamId(streamId);
+        assertThat(count, is(1));
 
+        final Optional<StreamStatus> result = jdbcRepository.findByStreamIdAndSource(streamId, "sjp");
+
+        assertThat(result.get().getSource(), is("sjp"));
+        assertThat(result.get().getVersion(), is(2L));
+    }
 
     @Test
     public void shouldUpdateSourceWhenUnknown() throws Exception {
@@ -187,4 +206,20 @@ private StreamStatus streamStatusOf(final UUID id, final Long version, final Str
         return new StreamStatus(id, version, source);
     }
 
+    private long initialiseBuffer(final UUID streamId, final String source) {
+        jdbcRepository.updateSource(streamId,source);
+        final Optional<StreamStatus> currentStatus = jdbcRepository.findByStreamIdAndSource(streamId, source);
+
+        if (!currentStatus.isPresent()) {
+            //this is to address race condition
+            //in case of primary key violation the exception gets thrown, event goes back into topic and the transaction gets retried
+            jdbcRepository
+                    .insert(new StreamStatus(streamId, INITIAL_VERSION, source));
+            return INITIAL_VERSION;
+
+        } else {
+            return currentStatus.get().getVersion();
+        }
+    }
+
 }