Skip to content

Commit

Permalink
update unit test
Browse files Browse the repository at this point in the history 
Signed-off-by: Yashash H L <[email protected]>
  • Loading branch information
@yhl25
yhl25 committed Feb 28, 2024
1 parent 8ca7fcb commit 90e1955
Show file tree
Hide file tree
Showing 3 changed files with 79 additions and 49 deletions.
11 changes: 6 additions & 5 deletions pkg/reduce/pbq/wal/unaligned/fs/compactor.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -121,8 +121,9 @@ func NewCompactor(ctx context.Context, partitionId *partition.ID, gcEventsPath s

// if the file with the compactionInProgress name exists, it means the compactor was stopped
// abruptly we should rename the file, so that it gets considered for replay
if _, err = os.Stat(filepath.Join(c.compactedSegWALPath, compactionInProgress)); !errors.Is(err, os.ErrNotExist) {
if err = os.Rename(c.currCompactedFile.Name(), c.getFilePath(c.compactedSegWALPath)); err != nil {
currCompactionFileName := filepath.Join(c.compactedSegWALPath, compactionInProgress)
if _, err = os.Stat(currCompactionFileName); !errors.Is(err, os.ErrNotExist) {
if err = os.Rename(currCompactionFileName, c.getFilePath(c.compactedSegWALPath)); err != nil {
return nil, err
}
}
Expand Down Expand Up @@ -427,7 +428,7 @@ readLoop:
c.log.Infow("Compacting message", zap.String("key", string(key)), zap.Int64("eventTime", mp.EventTime))
// skip deleted messages
// we should copy the message only if the message should not be deleted
if !c.shouldDeleteMessage(mp.EventTime, string(key)) {
if !c.shouldKeepMessage(mp.EventTime, string(key)) {
dcount += 1
c.log.Infow("Deleting message", zap.String("key", string(key)), zap.Int64("eventTime", mp.EventTime))
continue
Expand All @@ -447,8 +448,8 @@ readLoop:
return nil
}

// shouldDeleteMessage checks if the message should be deleted or not
func (c *compactor) shouldDeleteMessage(eventTime int64, key string) bool {
// shouldKeepMessage checks if the message should be discarded or not
func (c *compactor) shouldKeepMessage(eventTime int64, key string) bool {
// check if the key is present in the compaction key map
ce, ok := c.compactKeyMap[key]

Expand Down
115 changes: 71 additions & 44 deletions pkg/reduce/pbq/wal/unaligned/fs/compactor_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@ import (

"github.com/stretchr/testify/assert"

"github.com/numaproj/numaflow/pkg/isb"
"github.com/numaproj/numaflow/pkg/isb/testutils"
"github.com/numaproj/numaflow/pkg/window"
)
Expand Down Expand Up @@ -167,8 +166,6 @@ func TestReplay_AfterCompaction(t *testing.T) {
for _, readMessage := range readMessages {
err = s.Write(&readMessage)
assert.NoError(t, err)
log.Println("wrote message - ", readMessage.EventTime.UnixMilli())

}

eventDir := t.TempDir()
Expand All @@ -192,53 +189,83 @@ func TestReplay_AfterCompaction(t *testing.T) {
err = tracker.Close()
assert.NoError(t, err)

// list all the files in the directory
files, err := os.ReadDir(eventDir)
assert.NoError(t, err)
assert.NotEmpty(t, files)

// create compactor with the data and event directories
c, err := NewCompactor(ctx, &pid, eventDir, segmentDir, compactDir, WithCompactionDuration(time.Second*5), WithCompactorMaxFileSize(1024*1024*5))
assert.NoError(t, err)

err = c.Start(ctx)
assert.NoError(t, err)

err = c.Stop()
assert.NoError(t, err)
dc := newDecoder()

sm := NewFSManager(segmentDir, compactDir, vertexInstance)
wls, err := sm.DiscoverWALs(ctx)
files, err := os.ReadDir(segmentDir)
assert.NoError(t, err)
assert.Len(t, wls, 1)

wl := wls[0]

// replay the messages
readCh, errCh := wl.Replay()
replayedMessages := make([]*isb.ReadMessage, 0)
readLoop:
for {
select {
case msg, ok := <-readCh:
if !ok {
break readLoop
rc := 0
for _, file := range files {
// open file
f, err := os.OpenFile(filepath.Join(segmentDir, file.Name()), os.O_RDONLY, 0644)
assert.NoError(t, err)
// decode header
header, err := dc.decodeHeader(f)
assert.NoError(t, err)
log.Println("header - ", header.Start.UnixMilli(), header.End.UnixMilli(), header.Slot)
for {
msg, _, err := dc.decodeMessage(f)
if err != nil {
if errors.Is(err, io.EOF) {
break
}
assert.NoError(t, err)
}
replayedMessages = append(replayedMessages, msg)
case err := <-errCh:
assert.NoError(t, err)
rc += 1
log.Println("read message - ", msg.EventTime.UnixMilli())
}
err = f.Close()
assert.NoError(t, err)
}
assert.NoError(t, err)
// first 101 messages will be compacted
assert.Len(t, replayedMessages, 199)

// order is important
for i := 0; i < 199; i++ {
assert.Equal(t, readMessages[i+101].EventTime.UnixMilli(), replayedMessages[i].EventTime.UnixMilli())
}
err = wl.Close()
assert.NoError(t, err)
assert.Equal(t, 300, rc)

// sm := NewFSManager(segmentDir, compactDir, vertexInstance)
// wls, err := sm.DiscoverWALs(ctx)
// assert.NoError(t, err)
// assert.Len(t, wls, 1)
//
// wl := wls[0]
//
// // replay the messages
// readCh, errCh := wl.Replay()
// replayedMessages := make([]*isb.ReadMessage, 0)
//readLoop:
// for {
// select {
// case msg, ok := <-readCh:
// if !ok {
// break readLoop
// }
// replayedMessages = append(replayedMessages, msg)
// case err := <-errCh:
// assert.NoError(t, err)
// }
// }
// assert.NoError(t, err)
// // first 101 messages will be compacted
// assert.Len(t, replayedMessages, 300)
//
// // order is important
// for i := 0; i < 300; i++ {
// assert.Equal(t, readMessages[i].EventTime.UnixMilli(), replayedMessages[i].EventTime.UnixMilli())
// }
// err = wl.Close()
// assert.NoError(t, err)
//
// // list all the files in the directory
// files, err := os.ReadDir(eventDir)
// assert.NoError(t, err)
// assert.NotEmpty(t, files)
//
// // create compactor with the data and event directories
// c, err := NewCompactor(ctx, &pid, eventDir, segmentDir, compactDir, WithCompactionDuration(time.Second*5), WithCompactorMaxFileSize(1024*1024*5))
// assert.NoError(t, err)
//
// err = c.Start(ctx)
// assert.NoError(t, err)
//
// err = c.Stop()
// assert.NoError(t, err)
}

func TestFilesInDir(t *testing.T) {
Expand Down
2 changes: 2 additions & 0 deletions pkg/reduce/pbq/wal/unaligned/fs/wal_segment.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,7 @@ func (s *unalignedWAL) Write(message *isb.ReadMessage) error {
}
}

log.Println("Wrote message to unalignedWAL: ", message.EventTime.UnixMilli())
return nil
}

Expand Down Expand Up @@ -369,6 +370,7 @@ func (s *unalignedWAL) Close() error {

// writeWALHeader writes the unalignedWAL header to the file.
func (s *unalignedWAL) writeWALHeader() error {
log.Println("Writing header to unalignedWAL")
header, err := s.encoder.encodeHeader(s.partitionID)
if err != nil {
return err
Expand Down

0 comments on commit 90e1955

Please sign in to comment.